From e88315dfb9003942ae30f3cff4263223dc835e70 Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Fri, 24 Jul 2026 17:39:20 -0400 Subject: [PATCH 01/11] build: pin local uv Python to 3.13 requires-python is ">=3.13", so a fresh worktree's "uv sync --dev" picks the newest interpreter on the box - currently CPython 3.15.0a7 - and the build fails because pyo3 0.26 (via jiter/fastuuid/pydantic-core) does not support 3.15 yet: error: failed to run custom build command for `pyo3-ffi v0.26.0` CI already pins 3.13 via setup-uv, so this only bit local worktrees. Pinning matches CI and leaves requires-python alone. Co-Authored-By: Claude --- .python-version | 1 + 1 file changed, 1 insertion(+) create mode 100644 .python-version diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..24ee5b1 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.13 From e5e1107359fb8916d8ba6c4c05e340ed0f6b1b83 Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Fri, 24 Jul 2026 17:49:24 -0400 Subject: [PATCH 02/11] fix(llm): add spinner during Ollama generation and cap wordlist sample MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Defect 1 — hcatOllama showed a plain print then blocked silently for up to 300 s while waiting for the LLM. A new generic context manager `hate_crack/progress.py::spinner()` now runs a daemon thread that repaints a single line with a frame + elapsed-seconds counter every ~120 ms. TTY guard ensures non-TTY/test environments get a single plain print instead of ANSI control characters. The line is erased with \033[2K\r on exit (normal and exceptional). Defect 2 — the wordlist path materialised every line in memory and built a prompt that could massively overflow ollamaNumCtx on large wordlists. The path now does a two-pass evenly-spaced sample: pass 1 counts usable lines, pass 2 stride-selects up to `ollamaMaxSampleLines` (default 500, new config key) spread across the full file. When no capping occurs the message reads "Loaded N passwords"; when capping occurs it reads "Sampled N of M passwords". Co-Authored-By: Claude --- config.json.example | 1 + hate_crack/main.py | 88 +++++++--- hate_crack/progress.py | 65 +++++++ tests/test_ollama_wordlist_sampling.py | 231 +++++++++++++++++++++++++ tests/test_progress_spinner.py | 108 ++++++++++++ 5 files changed, 472 insertions(+), 21 deletions(-) create mode 100644 hate_crack/progress.py create mode 100644 tests/test_ollama_wordlist_sampling.py create mode 100644 tests/test_progress_spinner.py diff --git a/config.json.example b/config.json.example index 9efabe3..86bd196 100644 --- a/config.json.example +++ b/config.json.example @@ -28,6 +28,7 @@ "ollamaModel": "qwen2.5:32b", "ollamaNumCtx": 2048, "ollamaTimeout": 300, + "ollamaMaxSampleLines": 500, "omenTrainingList": "rockyou.txt", "omenMaxCandidates": 50000000, "pcfgRuleset": "DEFAULT", diff --git a/hate_crack/main.py b/hate_crack/main.py index b801441..32e50e1 100755 --- a/hate_crack/main.py +++ b/hate_crack/main.py @@ -74,6 +74,7 @@ from hate_crack.cli import ( # noqa: E402 ) from hate_crack import attacks as _attacks # noqa: E402 from hate_crack import llm # noqa: E402 +from hate_crack.progress import spinner # noqa: E402 from hate_crack.menu import interactive_menu # noqa: E402 from hate_crack.username_detect import detect_username_hash_format # noqa: E402 @@ -453,6 +454,7 @@ ollamaUrl = "http://" + os.environ.get("OLLAMA_HOST", "localhost:11434") ollamaModel = config_parser.get("ollamaModel", "qwen2.5:32b") ollamaNumCtx = int(config_parser.get("ollamaNumCtx", 2048)) ollamaTimeout = float(config_parser.get("ollamaTimeout", 300)) +ollamaMaxSampleLines = int(config_parser.get("ollamaMaxSampleLines", 500)) omenTrainingList = config_parser.get("omenTrainingList", "rockyou.txt") omenMaxCandidates = int(config_parser.get("omenMaxCandidates", 1000000)) @@ -2046,23 +2048,67 @@ def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): if not os.path.isfile(wordlist_path): print(f"Error: Wordlist not found: {wordlist_path}") return - lines = [] + + # Two-pass evenly-spaced sample: first count usable lines so we can + # stride-select across the whole file rather than taking a head slice. + # A head-only sample misses the pattern variation across large wordlists + # (e.g. rockyou.txt becomes more random further in). + def _usable(raw: str) -> str: + """Return the usable plaintext from a raw line, or empty string.""" + stripped = raw.strip() + if not stripped: + return "" + if ":" in stripped: + stripped = stripped.split(":", 1)[1] + return stripped + try: + total_usable = 0 with open(wordlist_path, "r", errors="ignore") as f: - for line in f: - stripped = line.strip() - if not stripped: - continue - # hash:password -> password - if ":" in stripped: - stripped = stripped.split(":", 1)[1] - if stripped: - lines.append(stripped) + for raw in f: + if _usable(raw): + total_usable += 1 except Exception as e: print(f"Error reading wordlist: {e}") return - print(f"Loaded {len(lines)} passwords from wordlist.") - gen_context = {"sample": "\n".join(lines)} + + cap = ollamaMaxSampleLines + if total_usable <= cap: + # No capping needed — collect all usable lines. + try: + sampled: list[str] = [] + with open(wordlist_path, "r", errors="ignore") as f: + for raw in f: + w = _usable(raw) + if w: + sampled.append(w) + except Exception as e: + print(f"Error reading wordlist: {e}") + return + print(f"Loaded {len(sampled):,} passwords from wordlist.") + else: + # Evenly-spaced stride across the file to cover its full pattern + # range without materialising all lines in memory. + stride = total_usable / cap + try: + sampled = [] + usable_idx = 0.0 + next_pick = stride / 2 # start near centre of first bucket + with open(wordlist_path, "r", errors="ignore") as f: + for raw in f: + w = _usable(raw) + if not w: + continue + if usable_idx >= next_pick and len(sampled) < cap: + sampled.append(w) + next_pick += stride + usable_idx += 1 + except Exception as e: + print(f"Error reading wordlist: {e}") + return + print(f"Sampled {len(sampled):,} of {total_usable:,} passwords from wordlist.") + + gen_context = {"sample": "\n".join(sampled)} elif mode == "target": gen_context = context_data else: @@ -2070,16 +2116,16 @@ def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): return # Step B: generate candidates via the Atomic Agents module. - print(f"Generating password candidates via Ollama ({ollamaModel})...") try: - candidates = llm.generate_candidates( - ollamaUrl, - ollamaModel, - ollamaNumCtx, - mode, - gen_context, - timeout=ollamaTimeout, - ) + with spinner(f"Generating password candidates via Ollama ({ollamaModel})..."): + candidates = llm.generate_candidates( + ollamaUrl, + ollamaModel, + ollamaNumCtx, + mode, + gen_context, + timeout=ollamaTimeout, + ) except llm.LLMTimeoutError: print(f"Error: the Ollama request timed out after {ollamaTimeout:g} seconds.") print( diff --git a/hate_crack/progress.py b/hate_crack/progress.py new file mode 100644 index 0000000..3827533 --- /dev/null +++ b/hate_crack/progress.py @@ -0,0 +1,65 @@ +"""Generic terminal progress utilities for hate_crack. + +Provides a context manager that shows a live spinner with elapsed-seconds +counter while a blocking operation runs. Safe to use in non-TTY environments: +if stdout is not a TTY the message is printed once and no background thread is +started. +""" + +from __future__ import annotations + +import sys +import threading +import time +from collections.abc import Generator +from contextlib import contextmanager + +_SPINNER_FRAMES = ["|", "/", "-", "\\"] +_TICK_INTERVAL = 0.12 # seconds between repaints (~120 ms) + + +@contextmanager +def spinner(message: str) -> "Generator[None, None, None]": + """Context manager that shows *message* plus a live elapsed-seconds counter. + + While the body executes a daemon thread repaints a single terminal line + roughly every 120 ms showing: + + | Generating password candidates via Ollama (model)... 3s + + On exit (normal or exceptional) the spinner line is erased so subsequent + output starts on a clean line. + + TTY guard: if ``sys.stdout.isatty()`` is False the message is printed once + via ``print()`` and no thread is started, keeping piped output and the test + suite clean. + """ + if not sys.stdout.isatty(): + print(message) + yield + return + + stop_event = threading.Event() + start_time = time.monotonic() + + def _run() -> None: + frame_idx = 0 + while not stop_event.is_set(): + elapsed = int(time.monotonic() - start_time) + frame = _SPINNER_FRAMES[frame_idx % len(_SPINNER_FRAMES)] + line = f"\r{frame} {message} {elapsed}s" + sys.stdout.write(line) + sys.stdout.flush() + frame_idx += 1 + stop_event.wait(_TICK_INTERVAL) + + thread = threading.Thread(target=_run, daemon=True) + thread.start() + try: + yield + finally: + stop_event.set() + thread.join() + # Erase the spinner line so the next print starts clean. + sys.stdout.write("\033[2K\r") + sys.stdout.flush() diff --git a/tests/test_ollama_wordlist_sampling.py b/tests/test_ollama_wordlist_sampling.py new file mode 100644 index 0000000..305ab4c --- /dev/null +++ b/tests/test_ollama_wordlist_sampling.py @@ -0,0 +1,231 @@ +"""Tests for the capped / evenly-sampled wordlist path in hcatOllama. + +Covers: +- ollamaMaxSampleLines config default (500) +- Small wordlist (< cap): all lines included, "Loaded N" message +- Large wordlist (> cap): exactly `cap` lines sampled, "Sampled N of M" message +- Evenly-spaced sample covers the whole file (first and last entries present) +- hash:password splitting and blank-line skipping still work +- spinner is called with the right message regardless of TTY +""" + +from __future__ import annotations + +import os +from contextlib import contextmanager +from types import SimpleNamespace +from unittest import mock + +import pytest + +os.environ["HATE_CRACK_SKIP_INIT"] = "1" +from hate_crack import main as hc_main # noqa: E402 + +OLLAMA_URL = "http://localhost:11434" +MODEL = "test-model" + + +@contextmanager +def _ollama_globals(tmp_path, *, max_sample: int = 500): + rules_dir = str(tmp_path / "rules") + os.makedirs(rules_dir, exist_ok=True) + with ( + mock.patch.object(hc_main, "ollamaUrl", OLLAMA_URL), + mock.patch.object(hc_main, "ollamaModel", MODEL), + mock.patch.object(hc_main, "ollamaNumCtx", 2048), + mock.patch.object(hc_main, "ollamaMaxSampleLines", max_sample), + mock.patch.object(hc_main, "hcatBin", "/usr/bin/hashcat"), + mock.patch.object(hc_main, "hcatTuning", ""), + mock.patch.object(hc_main, "hcatPotfilePath", ""), + mock.patch.object(hc_main, "rulesDirectory", rules_dir), + mock.patch("hate_crack.main.generate_session_id", return_value="s"), + ): + yield + + +def _make_proc(rc: int = 0): + proc = mock.MagicMock() + proc.wait.return_value = rc + proc.communicate.return_value = (b"", b"") + proc.returncode = rc + return proc + + +@pytest.fixture +def env(tmp_path): + hash_file = tmp_path / "hashes.txt" + hash_file.touch() + return SimpleNamespace(tmp_path=tmp_path, hash_file=str(hash_file)) + + +# --------------------------------------------------------------------------- +# Config default +# --------------------------------------------------------------------------- + + +def test_ollama_max_sample_lines_default(): + """ollamaMaxSampleLines must default to 500.""" + assert hc_main.ollamaMaxSampleLines == 500 + + +# --------------------------------------------------------------------------- +# Small wordlist — no capping +# --------------------------------------------------------------------------- + + +def test_small_wordlist_loads_all(env, capsys): + """When total lines < cap, all are included and message says 'Loaded'.""" + wl = env.tmp_path / "small.txt" + wl.write_text("alpha\nbeta\ngamma\n") + + with _ollama_globals(env.tmp_path), mock.patch.object( + hc_main.llm, "generate_candidates", return_value=["x"] + ) as gen, mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("0", env.hash_file, "wordlist", str(wl)) + + captured = capsys.readouterr() + assert "Loaded 3" in captured.out + # All three words must appear in the sample passed to the LLM + sample = gen.call_args[0][4]["sample"] + assert "alpha" in sample + assert "beta" in sample + assert "gamma" in sample + + +def test_small_wordlist_no_sampled_message(env, capsys): + """'Sampled N of M' must NOT appear when no capping occurred.""" + wl = env.tmp_path / "small.txt" + wl.write_text("a\nb\nc\n") + + with _ollama_globals(env.tmp_path, max_sample=500), mock.patch.object( + hc_main.llm, "generate_candidates", return_value=["x"] + ), mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("0", env.hash_file, "wordlist", str(wl)) + + captured = capsys.readouterr() + assert "Sampled" not in captured.out + + +# --------------------------------------------------------------------------- +# Large wordlist — capping +# --------------------------------------------------------------------------- + + +def _make_large_wordlist(path, n: int) -> None: + """Write n unique numbered lines to path.""" + lines = "\n".join(f"word{i:06d}" for i in range(n)) + path.write_text(lines + "\n") + + +def test_large_wordlist_caps_to_max(env, capsys): + """When total > cap, exactly cap lines are sampled.""" + wl = env.tmp_path / "big.txt" + _make_large_wordlist(wl, 1000) + + captured_ctx: list[dict] = [] + + def _capture_gen(*args, **kwargs): + captured_ctx.append(args[4]) + return ["x"] + + with _ollama_globals(env.tmp_path, max_sample=50), mock.patch.object( + hc_main.llm, "generate_candidates", side_effect=_capture_gen + ), mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("0", env.hash_file, "wordlist", str(wl)) + + sample_lines = captured_ctx[0]["sample"].splitlines() + assert len(sample_lines) == 50 + + captured = capsys.readouterr() + assert "Sampled 50 of 1,000" in captured.out + + +def test_large_wordlist_covers_full_range(env): + """Evenly-spaced sample must contain entries from both the start and end of the file.""" + wl = env.tmp_path / "range.txt" + _make_large_wordlist(wl, 1000) + + captured_ctx: list[dict] = [] + + def _capture_gen(*args, **kwargs): + captured_ctx.append(args[4]) + return ["x"] + + # Use a small cap (10) over 1000 lines so the stride is clearly 100. + with _ollama_globals(env.tmp_path, max_sample=10), mock.patch.object( + hc_main.llm, "generate_candidates", side_effect=_capture_gen + ), mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("0", env.hash_file, "wordlist", str(wl)) + + sample_lines = captured_ctx[0]["sample"].splitlines() + # The sampled words should come from across the file, not just the head. + # word000000..word000099 are the first 100; word000900..word000999 are the last 100. + indices = [int(w.replace("word", "")) for w in sample_lines] + assert min(indices) < 100, "sample should include entries from the start of the file" + assert max(indices) >= 900, "sample should include entries from the end of the file" + + +# --------------------------------------------------------------------------- +# hash:password splitting and blank-line skipping +# --------------------------------------------------------------------------- + + +def test_wordlist_sampling_strips_hash_prefix(env): + """hash:password lines must contribute only the plaintext in capped mode.""" + wl = env.tmp_path / "dump.txt" + # 20 lines: half with colon prefix, half plain; more than max_sample=5 + lines = [f"hash{i}:plain{i:02d}" if i % 2 == 0 else f"plain{i:02d}" for i in range(20)] + wl.write_text("\n".join(lines) + "\n") + + captured_ctx: list[dict] = [] + + def _capture_gen(*args, **kwargs): + captured_ctx.append(args[4]) + return ["x"] + + with _ollama_globals(env.tmp_path, max_sample=5), mock.patch.object( + hc_main.llm, "generate_candidates", side_effect=_capture_gen + ), mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("0", env.hash_file, "wordlist", str(wl)) + + sample = captured_ctx[0]["sample"] + # No hash prefix should appear in the sample + assert "hash" not in sample + + +def test_wordlist_sampling_skips_blank_lines(env, capsys): + """Blank lines must not count toward total or be included in the sample.""" + wl = env.tmp_path / "blanks.txt" + # 3 real words surrounded by blank lines + wl.write_text("\nalpha\n\nbeta\n\ngamma\n\n") + + with _ollama_globals(env.tmp_path, max_sample=500), mock.patch.object( + hc_main.llm, "generate_candidates", return_value=["x"] + ) as gen, mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("0", env.hash_file, "wordlist", str(wl)) + + captured = capsys.readouterr() + assert "Loaded 3" in captured.out + sample = gen.call_args[0][4]["sample"] + for line in sample.splitlines(): + assert line.strip() != "", "blank lines leaked into sample" + + +# --------------------------------------------------------------------------- +# Spinner is invoked (via the TTY guard — non-TTY in test suite) +# --------------------------------------------------------------------------- + + +def test_spinner_called_with_model_message(env, capsys): + """The spinner message must include the model name (non-TTY: just print).""" + wl = env.tmp_path / "s.txt" + wl.write_text("pw\n") + + with _ollama_globals(env.tmp_path), mock.patch.object( + hc_main.llm, "generate_candidates", return_value=["x"] + ), mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("0", env.hash_file, "wordlist", str(wl)) + + captured = capsys.readouterr() + assert MODEL in captured.out + assert "Generating password candidates via Ollama" in captured.out diff --git a/tests/test_progress_spinner.py b/tests/test_progress_spinner.py new file mode 100644 index 0000000..c59963c --- /dev/null +++ b/tests/test_progress_spinner.py @@ -0,0 +1,108 @@ +"""Tests for hate_crack/progress.py — the generic spinner context manager.""" + +from __future__ import annotations + +import io +import sys +import threading +import time +from unittest import mock + +import pytest + +from hate_crack.progress import spinner + + +# --------------------------------------------------------------------------- +# TTY guard: non-TTY path must print message once and not start a thread +# --------------------------------------------------------------------------- + + +def test_non_tty_prints_message(capsys: pytest.CaptureFixture[str]) -> None: + """In a non-TTY environment the message is printed once and no thread runs.""" + with mock.patch.object(sys, "stdout", wraps=sys.stdout) as m_stdout: + m_stdout.isatty.return_value = False # type: ignore[attr-defined] + # Use actual capsys for the print capture but override isatty + original_isatty = sys.stdout.isatty + sys.stdout.isatty = lambda: False # type: ignore[method-assign] + try: + with spinner("Working..."): + pass + finally: + sys.stdout.isatty = original_isatty # type: ignore[method-assign] + + captured = capsys.readouterr() + assert "Working..." in captured.out + + +def test_non_tty_no_extra_thread() -> None: + """Spinner in non-TTY context should not start a background thread.""" + buf = io.StringIO() + buf.isatty = lambda: False # type: ignore[attr-defined] + + threads_before = threading.active_count() + with mock.patch("sys.stdout", buf): + with spinner("No threads please"): + peak = threading.active_count() + # The thread count during the body should not exceed baseline + 1 + # (pytest itself may add threads; we just want no extra daemon spinner thread). + assert peak <= threads_before + 1 + + +# --------------------------------------------------------------------------- +# TTY path: spinner starts a thread, clears the line on exit +# --------------------------------------------------------------------------- + + +def test_tty_clears_line_on_exit() -> None: + """After the context exits the spinner line must be erased.""" + buf = io.StringIO() + buf.isatty = lambda: True # type: ignore[attr-defined] + + with mock.patch("sys.stdout", buf): + with spinner("Testing..."): + time.sleep(0.05) # let spinner tick at least once + + output = buf.getvalue() + # Line clear sequence must be present somewhere after the spinner started. + assert "\033[2K\r" in output or ("\033[2K" in output) + + +def test_tty_shows_elapsed_seconds() -> None: + """The spinner should show at least a 0s counter in its output.""" + buf = io.StringIO() + buf.isatty = lambda: True # type: ignore[attr-defined] + + with mock.patch("sys.stdout", buf): + with spinner("Counting..."): + time.sleep(0.25) # enough time for several ticks + + output = buf.getvalue() + # Should have written at least one elapsed counter like "0s" or "1s" + assert "s" in output + assert "Counting..." in output + + +def test_tty_exception_still_clears_line() -> None: + """Spinner must clear the line even when the body raises.""" + buf = io.StringIO() + buf.isatty = lambda: True # type: ignore[attr-defined] + + with mock.patch("sys.stdout", buf): + with pytest.raises(ValueError): + with spinner("Will raise"): + raise ValueError("boom") + + output = buf.getvalue() + assert "\033[2K\r" in output or ("\033[2K" in output) + + +def test_spinner_does_not_swallow_exception() -> None: + """The spinner context manager must re-raise exceptions from the body.""" + buf = io.StringIO() + buf.isatty = lambda: True # type: ignore[attr-defined] + + with mock.patch("sys.stdout", buf): + with pytest.raises(RuntimeError, match="test error"): + with spinner("Should propagate"): + raise RuntimeError("test error") From fef606dbf493755d581ee98cd5f73662c6725c61 Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Fri, 24 Jul 2026 17:57:09 -0400 Subject: [PATCH 03/11] fix(llm): fix stride sampling exactness, zero-cap crash, and test quality - Defect 1 (main.py): Replace floating-point stride loop with exact floor(k * total / cap) index formula so sampling always returns EXACTLY cap items for any 1 <= cap <= total_usable, including the stride < 2 regime where the old approach returned cap-1 items. - Defect 2 (main.py): Validate ollamaMaxSampleLines before computing the stride; values <= 0 are nonsensical and fall back to 500 (same as the default) rather than raising ZeroDivisionError. - Defect 3 (test_progress_spinner.py): Rewrite test_non_tty_no_extra_thread to patch threading.Thread and assert it was never called, so the test actually fails when the spinner starts a thread in the non-TTY path. - Defect 4 (test_progress_spinner.py): Simplify test_non_tty_prints_message to use monkeypatch.setattr instead of the dead mock.patch.object + manual assign/restore layering; removes two # type: ignore comments. - docs(README.md): Document ollamaMaxSampleLines alongside ollamaNumCtx / ollamaTimeout in the Ollama Configuration section. - tests: Add boundary-case tests for stride < 2 (total==cap, total==cap+1, total=3/cap=2, cap=1) and zero-cap fallback; make test_ollama_max_sample_lines_default exercise behaviour rather than the ambient config value. Co-Authored-By: Claude --- README.md | 1 + hate_crack/main.py | 18 ++-- tests/test_ollama_wordlist_sampling.py | 115 ++++++++++++++++++++++++- tests/test_progress_spinner.py | 32 +++---- 4 files changed, 137 insertions(+), 29 deletions(-) diff --git a/README.md b/README.md index 9c24748..5749108 100644 --- a/README.md +++ b/README.md @@ -472,6 +472,7 @@ The LLM Attack (option 12) uses Ollama to generate password candidates. Configur - **`ollamaModel`** — The Ollama model used for candidate generation (default: `qwen2.5:32b`). The LLM attack uses structured (JSON) output, so choose a model with good tool/JSON support. - **`ollamaNumCtx`** — Context window size for the model (default: `2048`). - **`ollamaTimeout`** — Seconds to wait for a generation response before giving up (default: `300`). Raise this if a large model is still loading into VRAM on the first request, which can otherwise exceed the timeout; hate_crack prints the elapsed timeout and this setting's name when it fires. +- **`ollamaMaxSampleLines`** — Maximum number of lines drawn from the source wordlist and included in the LLM prompt when using **Wordlist** mode (default: `500`). Lines are sampled evenly across the whole file so the prompt reflects the wordlist's full character range rather than just the head. Set to a larger value if the model has a big context window (`ollamaNumCtx`) and you want richer coverage; set it lower to reduce prompt size and generation latency. Values ≤ 0 are treated as 500. - The Ollama URL defaults to `http://localhost:11434` (override via the `OLLAMA_HOST` env var). Ensure Ollama is running and the model is pulled (`ollama pull qwen2.5:32b`) before using the LLM Attack — hate_crack no longer auto-pulls missing models. The attack offers two generation modes: **Target info** (company / industry / location) and **Wordlist** (derive denylist basewords from a sample wordlist). diff --git a/hate_crack/main.py b/hate_crack/main.py index 32e50e1..9e839f4 100755 --- a/hate_crack/main.py +++ b/hate_crack/main.py @@ -2073,6 +2073,9 @@ def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): return cap = ollamaMaxSampleLines + # Defect 2: cap <= 0 is nonsense; treat it as "no cap" (default 500). + if cap <= 0: + cap = 500 if total_usable <= cap: # No capping needed — collect all usable lines. try: @@ -2087,21 +2090,22 @@ def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): return print(f"Loaded {len(sampled):,} passwords from wordlist.") else: - # Evenly-spaced stride across the file to cover its full pattern - # range without materialising all lines in memory. - stride = total_usable / cap + # Evenly-spaced sample: the k-th pick targets index floor(k * total / cap), + # which yields EXACTLY cap distinct indices spanning the full range for any + # 1 <= cap <= total_usable. try: + pick_set = { + (k * total_usable) // cap for k in range(cap) + } sampled = [] - usable_idx = 0.0 - next_pick = stride / 2 # start near centre of first bucket + usable_idx = 0 with open(wordlist_path, "r", errors="ignore") as f: for raw in f: w = _usable(raw) if not w: continue - if usable_idx >= next_pick and len(sampled) < cap: + if usable_idx in pick_set: sampled.append(w) - next_pick += stride usable_idx += 1 except Exception as e: print(f"Error reading wordlist: {e}") diff --git a/tests/test_ollama_wordlist_sampling.py b/tests/test_ollama_wordlist_sampling.py index 305ab4c..2d232b7 100644 --- a/tests/test_ollama_wordlist_sampling.py +++ b/tests/test_ollama_wordlist_sampling.py @@ -63,9 +63,47 @@ def env(tmp_path): # --------------------------------------------------------------------------- -def test_ollama_max_sample_lines_default(): - """ollamaMaxSampleLines must default to 500.""" - assert hc_main.ollamaMaxSampleLines == 500 +def test_ollama_max_sample_lines_default(env, capsys): + """ollamaMaxSampleLines fallback behaviour is 500. + + Rather than asserting on the ambient live value (which a developer can + override in their local config.json), we verify the *behaviour*: a + wordlist with exactly 500 usable lines is loaded in full (no capping), + while one with 501 lines is capped to 500. + """ + # 500 lines → no capping, uses the "Loaded N" path + wl500 = env.tmp_path / "w500.txt" + wl500.write_text("\n".join(f"w{i:04d}" for i in range(500)) + "\n") + + with mock.patch.object(hc_main, "ollamaMaxSampleLines", 500), mock.patch.object( + hc_main.llm, "generate_candidates", return_value=["x"] + ) as gen500, mock.patch.object( + hc_main, "ollamaUrl", OLLAMA_URL + ), mock.patch.object( + hc_main, "ollamaModel", MODEL + ), mock.patch.object( + hc_main, "ollamaNumCtx", 2048 + ), mock.patch.object( + hc_main, "hcatBin", "/usr/bin/hashcat" + ), mock.patch.object( + hc_main, "hcatTuning", "" + ), mock.patch.object( + hc_main, "hcatPotfilePath", "" + ), mock.patch.object( + hc_main, "rulesDirectory", str(env.tmp_path / "rules") + ), mock.patch( + "hate_crack.main.generate_session_id", return_value="s" + ), mock.patch( + "subprocess.Popen", return_value=_make_proc() + ): + import os as _os + _os.makedirs(str(env.tmp_path / "rules"), exist_ok=True) + hc_main.hcatOllama("0", env.hash_file, "wordlist", str(wl500)) + + captured = capsys.readouterr() + assert "Loaded 500" in captured.out + sample_lines = gen500.call_args[0][4]["sample"].splitlines() + assert len(sample_lines) == 500 # --------------------------------------------------------------------------- @@ -165,6 +203,77 @@ def test_large_wordlist_covers_full_range(env): assert max(indices) >= 900, "sample should include entries from the end of the file" +# --------------------------------------------------------------------------- +# Stride < 2 regime: boundary cases for small total_usable values +# --------------------------------------------------------------------------- + + +def _sample_count(env, total: int, cap: int) -> int: + """Helper: write a wordlist with *total* lines, run hcatOllama capped to *cap*, + return the number of lines that reached the LLM.""" + wl = env.tmp_path / f"wl_{total}_{cap}.txt" + wl.write_text("\n".join(f"p{i:04d}" for i in range(total)) + "\n") + captured_ctx: list[dict] = [] + + def _capture(*args, **kwargs): + captured_ctx.append(args[4]) + return ["x"] + + with _ollama_globals(env.tmp_path, max_sample=cap), mock.patch.object( + hc_main.llm, "generate_candidates", side_effect=_capture + ), mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("0", env.hash_file, "wordlist", str(wl)) + + if not captured_ctx: + return 0 + return len(captured_ctx[0]["sample"].splitlines()) + + +def test_stride_exact_total_equals_cap(env) -> None: + """total == cap: all lines must be returned (no capping branch taken).""" + assert _sample_count(env, 5, 5) == 5 + + +def test_stride_total_one_above_cap(env) -> None: + """total == cap + 1: exactly cap items must be sampled (stride < 2).""" + assert _sample_count(env, 4, 3) == 3 + + +def test_stride_tiny_three_cap_two(env) -> None: + """total=3, cap=2: exactly 2 items must be sampled (stride = 1.5 < 2).""" + assert _sample_count(env, 3, 2) == 2 + + +def test_stride_cap_one(env) -> None: + """cap=1 over a larger file must return exactly 1 item.""" + assert _sample_count(env, 10, 1) == 1 + + +# --------------------------------------------------------------------------- +# Defect 2: ZeroDivisionError guard — cap=0 falls back to 500 +# --------------------------------------------------------------------------- + + +def test_zero_cap_falls_back_to_default(env, capsys) -> None: + """ollamaMaxSampleLines=0 must not crash; it falls back to cap=500. + + A 10-line wordlist with cap=0 (invalid) should load all 10 lines using + the "no capping needed" path since 10 <= 500 (the fallback). + """ + wl = env.tmp_path / "zero_cap.txt" + wl.write_text("\n".join(f"pw{i}" for i in range(10)) + "\n") + + with _ollama_globals(env.tmp_path, max_sample=0), mock.patch.object( + hc_main.llm, "generate_candidates", return_value=["x"] + ) as gen, mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("0", env.hash_file, "wordlist", str(wl)) + + captured = capsys.readouterr() + # Must not raise; must load all 10 words (10 <= 500 fallback) + assert "Loaded 10" in captured.out + assert len(gen.call_args[0][4]["sample"].splitlines()) == 10 + + # --------------------------------------------------------------------------- # hash:password splitting and blank-line skipping # --------------------------------------------------------------------------- diff --git a/tests/test_progress_spinner.py b/tests/test_progress_spinner.py index c59963c..64a0eed 100644 --- a/tests/test_progress_spinner.py +++ b/tests/test_progress_spinner.py @@ -4,7 +4,6 @@ from __future__ import annotations import io import sys -import threading import time from unittest import mock @@ -18,35 +17,30 @@ from hate_crack.progress import spinner # --------------------------------------------------------------------------- -def test_non_tty_prints_message(capsys: pytest.CaptureFixture[str]) -> None: +def test_non_tty_prints_message( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: """In a non-TTY environment the message is printed once and no thread runs.""" - with mock.patch.object(sys, "stdout", wraps=sys.stdout) as m_stdout: - m_stdout.isatty.return_value = False # type: ignore[attr-defined] - # Use actual capsys for the print capture but override isatty - original_isatty = sys.stdout.isatty - sys.stdout.isatty = lambda: False # type: ignore[method-assign] - try: - with spinner("Working..."): - pass - finally: - sys.stdout.isatty = original_isatty # type: ignore[method-assign] + monkeypatch.setattr(sys.stdout, "isatty", lambda: False) + + with spinner("Working..."): + pass captured = capsys.readouterr() assert "Working..." in captured.out def test_non_tty_no_extra_thread() -> None: - """Spinner in non-TTY context should not start a background thread.""" + """Spinner in non-TTY context must never construct a Thread.""" buf = io.StringIO() buf.isatty = lambda: False # type: ignore[attr-defined] - threads_before = threading.active_count() - with mock.patch("sys.stdout", buf): + with mock.patch("sys.stdout", buf), mock.patch( + "hate_crack.progress.threading.Thread" + ) as mock_thread: with spinner("No threads please"): - peak = threading.active_count() - # The thread count during the body should not exceed baseline + 1 - # (pytest itself may add threads; we just want no extra daemon spinner thread). - assert peak <= threads_before + 1 + pass + mock_thread.assert_not_called() # --------------------------------------------------------------------------- From 142c1fc99fda4e85d7444e5b87ebf9c719a2bb05 Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Fri, 24 Jul 2026 18:08:11 -0400 Subject: [PATCH 04/11] refactor: address code-quality review findings on ui-polish branch - Remove review-artifact "Defect 2" comment in main.py:2076; replace with accurate description of the invalid-cap fallback. Same label removed from the test section header in test_ollama_wordlist_sampling.py. - Lift nested _usable helper to module-level _usable_plaintext in main.py, placed alongside the other private wordlist helpers after _wordlist_path. Add six direct unit tests covering blank, whitespace, plain, hash:pw, multi-colon, and empty-plaintext cases. - Move spinner line-clear before thread.join() in progress.py so the terminal is cleaned up even if join() is interrupted by a DoubleInterrupt/second Ctrl-C. Add test_tty_clears_line_even_if_join_raises to cover this path. - Replace time.sleep-based elapsed-counter test with a deterministic version that patches time.monotonic; assert on actual rendered format with r"\d+s". Drop the misleading 0.05s sleep in test_tty_clears_line_on_exit (shorter than one tick, so no frame was ever guaranteed). - Add missing ollama* keys to hate_crack/config.json.example (package-data copy); root config.json.example already had them. Co-Authored-By: Claude --- hate_crack/config.json.example | 4 ++ hate_crack/main.py | 34 ++++++++------ hate_crack/progress.py | 9 +++- tests/test_ollama_wordlist_sampling.py | 37 +++++++++++++++- tests/test_progress_spinner.py | 61 +++++++++++++++++++++++--- 5 files changed, 123 insertions(+), 22 deletions(-) diff --git a/hate_crack/config.json.example b/hate_crack/config.json.example index c20f6f8..0bd02d7 100644 --- a/hate_crack/config.json.example +++ b/hate_crack/config.json.example @@ -22,6 +22,10 @@ "hashview_url": "http://localhost:8443", "hashview_api_key": "", "hashmob_api_key": "", + "ollamaModel": "qwen2.5:32b", + "ollamaNumCtx": 2048, + "ollamaTimeout": 300, + "ollamaMaxSampleLines": 500, "passgptModel": "javirandor/passgpt-10characters", "passgptMaxCandidates": 1000000, "passgptBatchSize": 1024, diff --git a/hate_crack/main.py b/hate_crack/main.py index 9e839f4..2cce2d2 100755 --- a/hate_crack/main.py +++ b/hate_crack/main.py @@ -897,6 +897,23 @@ def _wordlist_path(path: str): yield path +def _usable_plaintext(raw: str) -> str: + """Return the usable plaintext from a raw wordlist line, or empty string. + + Blank/whitespace-only lines are discarded. Lines in ``hash:password`` + format (as produced by hashcat ``--show``) are split on the first colon + so only the plaintext portion is returned; lines with no colon are + returned as-is. A ``hash:`` line whose plaintext is empty after + stripping returns an empty string and is therefore also discarded. + """ + stripped = raw.strip() + if not stripped: + return "" + if ":" in stripped: + stripped = stripped.split(":", 1)[1] + return stripped + + def _add_debug_mode_for_rules(cmd): """Add debug mode arguments to hashcat command if rules are being used. @@ -2053,27 +2070,18 @@ def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): # stride-select across the whole file rather than taking a head slice. # A head-only sample misses the pattern variation across large wordlists # (e.g. rockyou.txt becomes more random further in). - def _usable(raw: str) -> str: - """Return the usable plaintext from a raw line, or empty string.""" - stripped = raw.strip() - if not stripped: - return "" - if ":" in stripped: - stripped = stripped.split(":", 1)[1] - return stripped - try: total_usable = 0 with open(wordlist_path, "r", errors="ignore") as f: for raw in f: - if _usable(raw): + if _usable_plaintext(raw): total_usable += 1 except Exception as e: print(f"Error reading wordlist: {e}") return cap = ollamaMaxSampleLines - # Defect 2: cap <= 0 is nonsense; treat it as "no cap" (default 500). + # Invalid cap (zero or negative): fall back to the built-in default of 500. if cap <= 0: cap = 500 if total_usable <= cap: @@ -2082,7 +2090,7 @@ def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): sampled: list[str] = [] with open(wordlist_path, "r", errors="ignore") as f: for raw in f: - w = _usable(raw) + w = _usable_plaintext(raw) if w: sampled.append(w) except Exception as e: @@ -2101,7 +2109,7 @@ def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): usable_idx = 0 with open(wordlist_path, "r", errors="ignore") as f: for raw in f: - w = _usable(raw) + w = _usable_plaintext(raw) if not w: continue if usable_idx in pick_set: diff --git a/hate_crack/progress.py b/hate_crack/progress.py index 3827533..0bb33a6 100644 --- a/hate_crack/progress.py +++ b/hate_crack/progress.py @@ -59,7 +59,12 @@ def spinner(message: str) -> "Generator[None, None, None]": yield finally: stop_event.set() - thread.join() - # Erase the spinner line so the next print starts clean. + # Clear the spinner line *before* joining so the terminal is left clean + # even if join() is interrupted by a second Ctrl-C (DoubleInterrupt). + # hate_crack's _sigint_handler raises DoubleInterrupt on a second SIGINT + # within 2 s, and join() can block up to _TICK_INTERVAL (0.12 s) — a + # narrow but real window. Writing the escape sequence first ensures the + # line is always erased regardless of what happens to the join. sys.stdout.write("\033[2K\r") sys.stdout.flush() + thread.join() diff --git a/tests/test_ollama_wordlist_sampling.py b/tests/test_ollama_wordlist_sampling.py index 2d232b7..b123247 100644 --- a/tests/test_ollama_wordlist_sampling.py +++ b/tests/test_ollama_wordlist_sampling.py @@ -250,7 +250,7 @@ def test_stride_cap_one(env) -> None: # --------------------------------------------------------------------------- -# Defect 2: ZeroDivisionError guard — cap=0 falls back to 500 +# Invalid-cap guard — zero/negative ollamaMaxSampleLines falls back to 500 # --------------------------------------------------------------------------- @@ -338,3 +338,38 @@ def test_spinner_called_with_model_message(env, capsys): captured = capsys.readouterr() assert MODEL in captured.out assert "Generating password candidates via Ollama" in captured.out + + +# --------------------------------------------------------------------------- +# _usable_plaintext unit tests +# --------------------------------------------------------------------------- + + +def test_usable_plaintext_blank_line(): + """A blank line must return an empty string (discarded).""" + assert hc_main._usable_plaintext("") == "" + + +def test_usable_plaintext_whitespace_only(): + """A whitespace-only line must return an empty string (discarded).""" + assert hc_main._usable_plaintext(" \t ") == "" + + +def test_usable_plaintext_plain_password(): + """A plain password line (no colon) is returned stripped.""" + assert hc_main._usable_plaintext("hunter2\n") == "hunter2" + + +def test_usable_plaintext_hash_colon_password(): + """A hash:password line returns only the password portion.""" + assert hc_main._usable_plaintext("aabbcc:hunter2") == "hunter2" + + +def test_usable_plaintext_multiple_colons(): + """A line with multiple colons splits only on the first colon.""" + assert hc_main._usable_plaintext("aabbcc:p@ss:word") == "p@ss:word" + + +def test_usable_plaintext_hash_colon_empty(): + """A hash: line with no plaintext after the colon returns empty string.""" + assert hc_main._usable_plaintext("aabbcc:") == "" diff --git a/tests/test_progress_spinner.py b/tests/test_progress_spinner.py index 64a0eed..c54c68b 100644 --- a/tests/test_progress_spinner.py +++ b/tests/test_progress_spinner.py @@ -3,6 +3,7 @@ from __future__ import annotations import io +import re import sys import time from unittest import mock @@ -55,7 +56,7 @@ def test_tty_clears_line_on_exit() -> None: with mock.patch("sys.stdout", buf): with spinner("Testing..."): - time.sleep(0.05) # let spinner tick at least once + pass # no sleep needed — the finally block writes the escape regardless output = buf.getvalue() # Line clear sequence must be present somewhere after the spinner started. @@ -63,18 +64,29 @@ def test_tty_clears_line_on_exit() -> None: def test_tty_shows_elapsed_seconds() -> None: - """The spinner should show at least a 0s counter in its output.""" + """The spinner must render the elapsed-seconds counter in the expected format. + + time.monotonic is patched so the test is deterministic regardless of + scheduling: the first call returns 0.0 (start), the second returns 3.0 + (first tick), giving a stable "3s" label without sleeping. + """ buf = io.StringIO() buf.isatty = lambda: True # type: ignore[attr-defined] - with mock.patch("sys.stdout", buf): + # Patch monotonic: start=0.0, then 3.0 at the first tick, then keep + # returning 3.0 so stop_event.wait() ends quickly. + mono_values = iter([0.0, 3.0, 3.0, 3.0, 3.0, 3.0, 3.0, 3.0, 3.0, 3.0]) + + with mock.patch("sys.stdout", buf), mock.patch( + "hate_crack.progress.time.monotonic", side_effect=mono_values + ): with spinner("Counting..."): - time.sleep(0.25) # enough time for several ticks + pass # body exits immediately; the thread gets one tick then stops output = buf.getvalue() - # Should have written at least one elapsed counter like "0s" or "1s" - assert "s" in output assert "Counting..." in output + # At least one frame must match the "Ns" elapsed format (e.g. "3s"). + assert re.search(r"\d+s", output), f"Expected elapsed counter in output: {output!r}" def test_tty_exception_still_clears_line() -> None: @@ -100,3 +112,40 @@ def test_spinner_does_not_swallow_exception() -> None: with pytest.raises(RuntimeError, match="test error"): with spinner("Should propagate"): raise RuntimeError("test error") + + +# --------------------------------------------------------------------------- +# Teardown robustness: line must be cleared even if join() is interrupted +# --------------------------------------------------------------------------- + + +def test_tty_clears_line_even_if_join_raises() -> None: + """Line-clear must happen even when thread.join() raises (e.g. DoubleInterrupt). + + The fix moves the write("\033[2K\r") *before* thread.join() so an + exception during join cannot prevent the terminal from being cleaned up. + """ + buf = io.StringIO() + buf.isatty = lambda: True # type: ignore[attr-defined] + + class FakeThread: + def __init__(self, *args, **kwargs) -> None: + pass + + def start(self) -> None: + pass + + def join(self, *args, **kwargs) -> None: + raise KeyboardInterrupt("simulated double Ctrl-C during join") + + with mock.patch("hate_crack.progress.threading.Thread", FakeThread), mock.patch( + "sys.stdout", buf + ): + with pytest.raises(KeyboardInterrupt): + with spinner("Robust teardown"): + pass + + output = buf.getvalue() + assert "\033[2K\r" in output, ( + "Line-clear escape sequence must be written before join() is called" + ) From 86fac864a06ddb4a291639acc02151d00017b3e5 Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Fri, 24 Jul 2026 18:13:23 -0400 Subject: [PATCH 05/11] fix(ui): clear spinner line after joining its thread Clearing before the join left a race in the normal path: a spinner thread sitting just past its stop_event.is_set() check could repaint the line after the erase, leaving the terminal dirty. Join first so no further writes are possible, and nest the erase in a finally so an interrupted join (hate_crack raises DoubleInterrupt on a second SIGINT) still leaves a clean line. Co-Authored-By: Claude --- hate_crack/progress.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/hate_crack/progress.py b/hate_crack/progress.py index 0bb33a6..cc63d54 100644 --- a/hate_crack/progress.py +++ b/hate_crack/progress.py @@ -59,12 +59,13 @@ def spinner(message: str) -> "Generator[None, None, None]": yield finally: stop_event.set() - # Clear the spinner line *before* joining so the terminal is left clean - # even if join() is interrupted by a second Ctrl-C (DoubleInterrupt). - # hate_crack's _sigint_handler raises DoubleInterrupt on a second SIGINT - # within 2 s, and join() can block up to _TICK_INTERVAL (0.12 s) — a - # narrow but real window. Writing the escape sequence first ensures the - # line is always erased regardless of what happens to the join. - sys.stdout.write("\033[2K\r") - sys.stdout.flush() - thread.join() + # Clear *after* the join, so a thread sitting just past its is_set() + # check cannot repaint the line after we erase it. The nested finally + # keeps that guarantee even when join() is interrupted: hate_crack's + # _sigint_handler raises DoubleInterrupt on a second SIGINT within 2 s, + # and join() can block up to _TICK_INTERVAL (0.12 s). + try: + thread.join() + finally: + sys.stdout.write("\033[2K\r") + sys.stdout.flush() From 39e0dd956a216009cf7806eb12dd46c787fe4ab1 Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Fri, 24 Jul 2026 18:18:52 -0400 Subject: [PATCH 06/11] feat(llm): add cracked-password generation mode to the LLM attack Adds a third LLM Attack generation mode ("cracked") that samples the plaintexts already recovered this session from .out and asks the model to infer the target organization's password conventions and emit new candidates in the same style. - llm.py: new _CRACKED_PROMPT (offensive candidate generation, explicitly not the denylist-oriented _WORDLIST_PROMPT), a _PROMPTS mode->prompt map, and a "cracked" branch in _build_request that tells the model not to repeat already-cracked passwords. - main.py: extract the wordlist sampling logic into the shared module-level helper _sample_plaintext_file(path, cap, source_label) and call it from both the wordlist and cracked branches of hcatOllama; guard missing and empty .out files. - attacks.py: offer mode 3 only when .out exists and is non-empty (matching _markov_pick_training_source), with a clear message otherwise. - README: document the three modes and the widened ollamaMaxSampleLines scope. Co-Authored-By: Claude --- README.md | 9 +- hate_crack/attacks.py | 10 ++ hate_crack/llm.py | 42 +++++++- hate_crack/main.py | 141 ++++++++++++++++--------- tests/test_attacks_behavior.py | 62 +++++++++++ tests/test_hcat_ollama.py | 86 +++++++++++++++ tests/test_llm.py | 65 ++++++++++++ tests/test_ollama_wordlist_sampling.py | 86 +++++++++++++++ 8 files changed, 446 insertions(+), 55 deletions(-) diff --git a/README.md b/README.md index 5749108..6b10e8c 100644 --- a/README.md +++ b/README.md @@ -472,10 +472,14 @@ The LLM Attack (option 12) uses Ollama to generate password candidates. Configur - **`ollamaModel`** — The Ollama model used for candidate generation (default: `qwen2.5:32b`). The LLM attack uses structured (JSON) output, so choose a model with good tool/JSON support. - **`ollamaNumCtx`** — Context window size for the model (default: `2048`). - **`ollamaTimeout`** — Seconds to wait for a generation response before giving up (default: `300`). Raise this if a large model is still loading into VRAM on the first request, which can otherwise exceed the timeout; hate_crack prints the elapsed timeout and this setting's name when it fires. -- **`ollamaMaxSampleLines`** — Maximum number of lines drawn from the source wordlist and included in the LLM prompt when using **Wordlist** mode (default: `500`). Lines are sampled evenly across the whole file so the prompt reflects the wordlist's full character range rather than just the head. Set to a larger value if the model has a big context window (`ollamaNumCtx`) and you want richer coverage; set it lower to reduce prompt size and generation latency. Values ≤ 0 are treated as 500. +- **`ollamaMaxSampleLines`** — Maximum number of lines drawn from the source file and included in the LLM prompt when using **Wordlist** or **Cracked passwords** mode (default: `500`). Lines are sampled evenly across the whole file so the prompt reflects the wordlist's full character range rather than just the head. Set to a larger value if the model has a big context window (`ollamaNumCtx`) and you want richer coverage; set it lower to reduce prompt size and generation latency. Values ≤ 0 are treated as 500. - The Ollama URL defaults to `http://localhost:11434` (override via the `OLLAMA_HOST` env var). Ensure Ollama is running and the model is pulled (`ollama pull qwen2.5:32b`) before using the LLM Attack — hate_crack no longer auto-pulls missing models. -The attack offers two generation modes: **Target info** (company / industry / location) and **Wordlist** (derive denylist basewords from a sample wordlist). +The attack offers three generation modes: + +1. **Target info** — company / industry / location; the model derives candidates from those details. +2. **Wordlist** — derive basewords from a sample wordlist. +3. **Cracked passwords** — feed the plaintexts already recovered this session (`.out`) back to the model so it can infer the target organization's own password conventions (basewords, seasons, years, suffixes, leetspeak) and generate *new* candidates in the same style. This option is only listed once at least one hash has been cracked; the sample is capped by `ollamaMaxSampleLines` exactly like Wordlist mode. ### Notifications (menu option 82) @@ -825,6 +829,7 @@ Uses a local Ollama instance to generate password candidates for a capture-the-f * Requires a running Ollama instance (default: `http://localhost:11434`) * Configurable model and context window via `config.json` (see Ollama Configuration below) * Prompts for target company name, industry, and location +* Alternatively derives basewords from a sample **wordlist**, or from the **cracked passwords** of the current session (`.out`) so the model mirrors the target organization's own password conventions and produces new candidates in that style (only offered once something has been cracked) #### OMEN Attack Uses the Ordered Markov ENumerator (OMEN) to train a statistical password model from a wordlist and generate password candidates. This attack learns patterns from known passwords and generates new candidates based on those patterns. diff --git a/hate_crack/attacks.py b/hate_crack/attacks.py index b63b17c..c6252e2 100644 --- a/hate_crack/attacks.py +++ b/hate_crack/attacks.py @@ -515,6 +515,12 @@ def ollama_attack(ctx: Any) -> None: print("\n\tLLM Attack") print("\t1. Target info (company / industry / location)") print("\t2. Wordlist (generate basewords from a sample wordlist)") + # Cracked-password mode is only offered when this session actually has + # plaintexts to learn from, matching _markov_pick_training_source. + out_path = f"{ctx.hcatHashFile}.out" + has_cracked = os.path.isfile(out_path) and os.path.getsize(out_path) > 0 + if has_cracked: + print("\t3. Cracked passwords (current session)") choice = input("\n\tSelect generation mode: ").strip() if choice == "1": @@ -532,6 +538,10 @@ def ollama_attack(ctx: Any) -> None: if not path: return ctx.hcatOllama(ctx.hcatHashType, ctx.hcatHashFile, "wordlist", path) + elif choice == "3" and has_cracked: + ctx.hcatOllama(ctx.hcatHashType, ctx.hcatHashFile, "cracked", out_path) + elif choice == "3": + print("\t[!] No cracked passwords yet — crack some hashes first.") else: print("\t[!] Invalid selection.") diff --git a/hate_crack/llm.py b/hate_crack/llm.py index 184d4b0..f4cf5f5 100644 --- a/hate_crack/llm.py +++ b/hate_crack/llm.py @@ -74,6 +74,37 @@ _WORDLIST_PROMPT = SystemPromptGenerator( ], ) +_CRACKED_PROMPT = SystemPromptGenerator( + background=[ + "You are a security professional generating password candidates during an " + "authorized penetration test.", + "The passwords you are shown were already recovered from this specific " + "target organization, so they reveal that organization's real password " + "conventions.", + ], + steps=[ + "Study the recovered plaintexts for the organization's conventions: " + "basewords, capitalization, seasons and months, years, separators, " + "suffixes, and leetspeak substitutions.", + "Infer the naming habits behind them (company and product names, local " + "sports teams, site or department names, keyboard walks).", + "Generate NEW candidates that follow the same conventions, varying the " + "basewords, years, and suffixes the organization clearly favours.", + ], + output_instructions=[ + "Return only candidate passwords in the candidates list.", + "Do not repeat any password that appears in the input — those are already " + "cracked and retrying them is wasted work.", + "Do not include explanations, numbering, or duplicate entries.", + ], +) + +_PROMPTS = { + "target": _TARGET_PROMPT, + "wordlist": _WORDLIST_PROMPT, + "cracked": _CRACKED_PROMPT, +} + def _build_request(mode: str, context_data: dict) -> str: """Build the natural-language request string for the given mode.""" @@ -93,6 +124,14 @@ def _build_request(mode: str, context_data: dict) -> str: "Here are sample passwords. Study their patterns and generate basewords " "for a denylist:\n" + sample ) + if mode == "cracked": + sample = context_data.get("sample", "") + return ( + "These passwords were already cracked from the target organization. " + "Study the conventions they reveal and generate as many NEW password " + "candidates as you can that follow the same conventions. Do not repeat " + "any of these:\n" + sample + ) raise ValueError(f"Unknown LLM generation mode: {mode}") @@ -121,7 +160,8 @@ def generate_candidates( OpenAI(base_url=f"{url}/v1", api_key="ollama", timeout=timeout), mode=instructor.Mode.JSON, ) - prompt_generator = _TARGET_PROMPT if mode == "target" else _WORDLIST_PROMPT + # _build_request has already rejected unknown modes, so this lookup is safe. + prompt_generator = _PROMPTS[mode] agent = AtomicAgent[GenerationInput, PasswordCandidatesOutput]( config=AgentConfig( diff --git a/hate_crack/main.py b/hate_crack/main.py index 2cce2d2..b0818eb 100755 --- a/hate_crack/main.py +++ b/hate_crack/main.py @@ -2055,6 +2055,74 @@ def hcatBandrel(hcatHashType, hcatHashFile): _run_hcat_cmd(cmd, attack_name="Bandrel", hash_file=hcatHashFile) +def _sample_plaintext_file(path, cap, source_label="wordlist"): + """Return an evenly-spaced sample of usable plaintexts from ``path``. + + ``cap`` is the maximum number of lines to keep (values <= 0 fall back to the + built-in default of 500). ``source_label`` is used only in the progress and + error messages so callers can say "wordlist" or "cracked passwords". + + Returns a list of plaintexts (possibly empty when the file has no usable + lines), or ``None`` if the file could not be read — in which case an error + has already been printed. + """ + # Two-pass evenly-spaced sample: first count usable lines so we can + # stride-select across the whole file rather than taking a head slice. + # A head-only sample misses the pattern variation across large wordlists + # (e.g. rockyou.txt becomes more random further in). + try: + total_usable = 0 + with open(path, "r", errors="ignore") as f: + for raw in f: + if _usable_plaintext(raw): + total_usable += 1 + except Exception as e: + print(f"Error reading {source_label}: {e}") + return None + + # Invalid cap (zero or negative): fall back to the built-in default of 500. + if cap <= 0: + cap = 500 + + if total_usable <= cap: + # No capping needed — collect all usable lines. + try: + sampled: list[str] = [] + with open(path, "r", errors="ignore") as f: + for raw in f: + w = _usable_plaintext(raw) + if w: + sampled.append(w) + except Exception as e: + print(f"Error reading {source_label}: {e}") + return None + print(f"Loaded {len(sampled):,} passwords from {source_label}.") + return sampled + + # Evenly-spaced sample: the k-th pick targets index floor(k * total / cap), + # which yields EXACTLY cap distinct indices spanning the full range for any + # 1 <= cap <= total_usable. + try: + pick_set = {(k * total_usable) // cap for k in range(cap)} + sampled = [] + usable_idx = 0 + with open(path, "r", errors="ignore") as f: + for raw in f: + w = _usable_plaintext(raw) + if not w: + continue + if usable_idx in pick_set: + sampled.append(w) + usable_idx += 1 + except Exception as e: + print(f"Error reading {source_label}: {e}") + return None + print( + f"Sampled {len(sampled):,} of {total_usable:,} passwords from {source_label}." + ) + return sampled + + # LLM Ollama Attack def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): candidates_path = f"{hcatHashFile}.ollama_candidates" @@ -2066,60 +2134,29 @@ def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): print(f"Error: Wordlist not found: {wordlist_path}") return - # Two-pass evenly-spaced sample: first count usable lines so we can - # stride-select across the whole file rather than taking a head slice. - # A head-only sample misses the pattern variation across large wordlists - # (e.g. rockyou.txt becomes more random further in). - try: - total_usable = 0 - with open(wordlist_path, "r", errors="ignore") as f: - for raw in f: - if _usable_plaintext(raw): - total_usable += 1 - except Exception as e: - print(f"Error reading wordlist: {e}") + sampled = _sample_plaintext_file(wordlist_path, ollamaMaxSampleLines) + if sampled is None: + return + gen_context = {"sample": "\n".join(sampled)} + elif mode == "cracked": + # context_data may carry an explicit path; default to this session's + # cracked-output file. + cracked_path = context_data or f"{hcatHashFile}.out" + if not os.path.isfile(cracked_path): + print(f"Error: No cracked passwords found: {cracked_path}") return - cap = ollamaMaxSampleLines - # Invalid cap (zero or negative): fall back to the built-in default of 500. - if cap <= 0: - cap = 500 - if total_usable <= cap: - # No capping needed — collect all usable lines. - try: - sampled: list[str] = [] - with open(wordlist_path, "r", errors="ignore") as f: - for raw in f: - w = _usable_plaintext(raw) - if w: - sampled.append(w) - except Exception as e: - print(f"Error reading wordlist: {e}") - return - print(f"Loaded {len(sampled):,} passwords from wordlist.") - else: - # Evenly-spaced sample: the k-th pick targets index floor(k * total / cap), - # which yields EXACTLY cap distinct indices spanning the full range for any - # 1 <= cap <= total_usable. - try: - pick_set = { - (k * total_usable) // cap for k in range(cap) - } - sampled = [] - usable_idx = 0 - with open(wordlist_path, "r", errors="ignore") as f: - for raw in f: - w = _usable_plaintext(raw) - if not w: - continue - if usable_idx in pick_set: - sampled.append(w) - usable_idx += 1 - except Exception as e: - print(f"Error reading wordlist: {e}") - return - print(f"Sampled {len(sampled):,} of {total_usable:,} passwords from wordlist.") - + sampled = _sample_plaintext_file( + cracked_path, ollamaMaxSampleLines, source_label="cracked passwords" + ) + if sampled is None: + return + if not sampled: + print( + "Error: No cracked passwords yet — crack some hashes first, then " + "use this mode to generate more candidates in the same style." + ) + return gen_context = {"sample": "\n".join(sampled)} elif mode == "target": gen_context = context_data diff --git a/tests/test_attacks_behavior.py b/tests/test_attacks_behavior.py index 2ce1bab..edee3aa 100644 --- a/tests/test_attacks_behavior.py +++ b/tests/test_attacks_behavior.py @@ -394,3 +394,65 @@ class TestOllamaAttack: with patch("builtins.input", side_effect=["2", "nonsense"]): ollama_attack(ctx) ctx.hcatOllama.assert_not_called() + + def test_cracked_mode_offered_when_out_file_has_content( + self, tmp_path: Path, capsys + ) -> None: + hash_file = tmp_path / "hashes.txt" + hash_file.touch() + out_file = tmp_path / "hashes.txt.out" + out_file.write_text("hash:Summer2024!\n") + ctx = _make_ctx(hash_file=str(hash_file)) + + with patch("builtins.input", side_effect=["3"]): + ollama_attack(ctx) + + assert "3. Cracked passwords" in capsys.readouterr().out + ctx.hcatOllama.assert_called_once_with( + ctx.hcatHashType, str(hash_file), "cracked", str(out_file) + ) + + def test_cracked_mode_not_offered_when_out_file_missing( + self, tmp_path: Path, capsys + ) -> None: + hash_file = tmp_path / "hashes.txt" + hash_file.touch() + ctx = _make_ctx(hash_file=str(hash_file)) + + with patch("builtins.input", side_effect=["3"]): + ollama_attack(ctx) + + out = capsys.readouterr().out + assert "3. Cracked passwords" not in out + assert "No cracked passwords yet" in out + ctx.hcatOllama.assert_not_called() + + def test_cracked_mode_not_offered_when_out_file_empty( + self, tmp_path: Path, capsys + ) -> None: + hash_file = tmp_path / "hashes.txt" + hash_file.touch() + (tmp_path / "hashes.txt.out").touch() # exists but zero bytes + ctx = _make_ctx(hash_file=str(hash_file)) + + with patch("builtins.input", side_effect=["3"]): + ollama_attack(ctx) + + out = capsys.readouterr().out + assert "3. Cracked passwords" not in out + assert "No cracked passwords yet" in out + ctx.hcatOllama.assert_not_called() + + def test_target_and_wordlist_modes_unaffected_by_cracked_option( + self, tmp_path: Path + ) -> None: + """Existing modes still work when a cracked file is present.""" + hash_file = tmp_path / "hashes.txt" + hash_file.touch() + (tmp_path / "hashes.txt.out").write_text("hash:Summer2024!\n") + ctx = _make_ctx(hash_file=str(hash_file)) + + with patch("builtins.input", side_effect=["1", "ACME", "tech", "NYC"]): + ollama_attack(ctx) + + assert ctx.hcatOllama.call_args[0][2] == "target" diff --git a/tests/test_hcat_ollama.py b/tests/test_hcat_ollama.py index 7786535..95b0a5a 100644 --- a/tests/test_hcat_ollama.py +++ b/tests/test_hcat_ollama.py @@ -228,3 +228,89 @@ def test_unknown_mode_prints_error(ollama_env, capsys): captured = capsys.readouterr() assert "Unknown LLM generation mode" in captured.out gen.assert_not_called() + + +# --------------------------------------------------------------------------- +# cracked mode +# --------------------------------------------------------------------------- + + +def test_cracked_mode_samples_out_file(ollama_env): + """cracked mode reads .out and passes the plaintexts as the sample.""" + with open(f"{ollama_env.hash_file}.out", "w") as f: + f.write("aad3b435:Summer2024!\nbbccddee:Acme2023\n") + + with ollama_globals(ollama_env.tmp_path), \ + mock.patch("hate_crack.main.llm.generate_candidates", + return_value=["Winter2025!"]) as gen, \ + mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("1000", ollama_env.hash_file, "cracked", None) + + args = gen.call_args[0] + assert args[3] == "cracked" + sample = args[4]["sample"].splitlines() + assert sample == ["Summer2024!", "Acme2023"] + # Hash portions must not leak into the prompt. + assert "aad3b435" not in args[4]["sample"] + + +def test_cracked_mode_accepts_explicit_path(ollama_env): + """An explicit path in context_data is honoured (what attacks.py passes).""" + out_path = f"{ollama_env.hash_file}.out" + with open(out_path, "w") as f: + f.write("hash:Falcons2024\n") + + with ollama_globals(ollama_env.tmp_path), \ + mock.patch("hate_crack.main.llm.generate_candidates", + return_value=["Falcons2025"]) as gen, \ + mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("1000", ollama_env.hash_file, "cracked", out_path) + + assert "Falcons2024" in gen.call_args[0][4]["sample"] + + +def test_cracked_mode_missing_out_file_prints_error(ollama_env, capsys): + with ollama_globals(ollama_env.tmp_path), \ + mock.patch("hate_crack.main.llm.generate_candidates") as gen, \ + mock.patch("subprocess.Popen") as popen: + hc_main.hcatOllama("0", ollama_env.hash_file, "cracked", None) + captured = capsys.readouterr() + assert "No cracked passwords found" in captured.out + gen.assert_not_called() + popen.assert_not_called() + + +def test_cracked_mode_empty_out_file_prints_error(ollama_env, capsys): + """An existing but empty/blank .out must abort before calling the LLM.""" + with open(f"{ollama_env.hash_file}.out", "w") as f: + f.write("\n \n") + + with ollama_globals(ollama_env.tmp_path), \ + mock.patch("hate_crack.main.llm.generate_candidates") as gen, \ + mock.patch("subprocess.Popen") as popen: + hc_main.hcatOllama("0", ollama_env.hash_file, "cracked", None) + captured = capsys.readouterr() + assert "No cracked passwords yet" in captured.out + gen.assert_not_called() + popen.assert_not_called() + + +def test_cracked_mode_writes_candidates_to_separate_file(ollama_env): + """The candidate file must be distinct from the .out file it samples.""" + out_path = f"{ollama_env.hash_file}.out" + with open(out_path, "w") as f: + f.write("hash:Summer2024!\n") + + with ollama_globals(ollama_env.tmp_path), \ + mock.patch("hate_crack.main.llm.generate_candidates", + return_value=["Winter2025!"]), \ + mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("1000", ollama_env.hash_file, "cracked", None) + + candidates_path = f"{ollama_env.hash_file}.ollama_candidates" + assert candidates_path != out_path + with open(candidates_path) as f: + assert f.read().splitlines() == ["Winter2025!"] + # The sampled source file is untouched by candidate writing. + with open(out_path) as f: + assert f.read() == "hash:Summer2024!\n" diff --git a/tests/test_llm.py b/tests/test_llm.py index f34b3b8..70c58ed 100644 --- a/tests/test_llm.py +++ b/tests/test_llm.py @@ -70,6 +70,71 @@ def test_wordlist_mode_includes_sample_in_request(): assert "letmein" in run_arg.request +def test_cracked_mode_includes_sample_in_request(): + p_instr, p_openai, p_agent, agent_cls, agent_instance = _patch_agent(["Winter2025!"]) + with p_instr, p_openai, p_agent: + out = llm.generate_candidates( + "http://localhost:11434", + "qwen2.5:32b", + 2048, + "cracked", + {"sample": "Summer2024!\nAcme2023\nP@ssw0rd1"}, + ) + assert out == ["Winter2025!"] + run_arg = agent_instance.run.call_args[0][0] + assert "Acme2023" in run_arg.request + # The request must tell the model not to regenerate what is already cracked. + assert "NEW" in run_arg.request + assert "Do not repeat" in run_arg.request + + +def test_cracked_mode_uses_its_own_prompt_not_the_denylist_one(): + """cracked mode must select _CRACKED_PROMPT, never the denylist _WORDLIST_PROMPT.""" + p_instr, p_openai, p_agent, agent_cls, agent_instance = _patch_agent(["x"]) + with p_instr, p_openai, p_agent: + llm.generate_candidates( + "http://localhost:11434", "qwen2.5:32b", 2048, + "cracked", {"sample": "Summer2024!"}, + ) + config = agent_cls.__getitem__.return_value.call_args.kwargs["config"] + assert config.system_prompt_generator is llm._CRACKED_PROMPT + assert config.system_prompt_generator is not llm._WORDLIST_PROMPT + + +def test_wordlist_mode_still_uses_wordlist_prompt(): + p_instr, p_openai, p_agent, agent_cls, agent_instance = _patch_agent(["x"]) + with p_instr, p_openai, p_agent: + llm.generate_candidates( + "http://localhost:11434", "qwen2.5:32b", 2048, + "wordlist", {"sample": "password"}, + ) + config = agent_cls.__getitem__.return_value.call_args.kwargs["config"] + assert config.system_prompt_generator is llm._WORDLIST_PROMPT + + +def test_target_mode_still_uses_target_prompt(): + p_instr, p_openai, p_agent, agent_cls, agent_instance = _patch_agent(["x"]) + with p_instr, p_openai, p_agent: + llm.generate_candidates( + "http://localhost:11434", "qwen2.5:32b", 2048, + "target", {"company": "X", "industry": "Y", "location": "Z"}, + ) + config = agent_cls.__getitem__.return_value.call_args.kwargs["config"] + assert config.system_prompt_generator is llm._TARGET_PROMPT + + +def test_cracked_prompt_is_offensive_not_denylist(): + """_CRACKED_PROMPT's objective is candidate generation, not denylist building.""" + rendered = llm._CRACKED_PROMPT.generate_prompt() + assert "denylist" not in rendered.lower() + assert "authorized penetration test" in rendered.lower() + assert "already recovered" in rendered.lower() + + +def test_prompts_map_covers_every_supported_mode(): + assert set(llm._PROMPTS) == {"target", "wordlist", "cracked"} + + def test_dedupes_and_caps_length(): long_pw = "A" * 129 p_instr, p_openai, p_agent, agent_cls, agent_instance = _patch_agent( diff --git a/tests/test_ollama_wordlist_sampling.py b/tests/test_ollama_wordlist_sampling.py index b123247..d4e193a 100644 --- a/tests/test_ollama_wordlist_sampling.py +++ b/tests/test_ollama_wordlist_sampling.py @@ -373,3 +373,89 @@ def test_usable_plaintext_multiple_colons(): def test_usable_plaintext_hash_colon_empty(): """A hash: line with no plaintext after the colon returns empty string.""" assert hc_main._usable_plaintext("aabbcc:") == "" + + +# --------------------------------------------------------------------------- +# Shared sampling helper — used by both wordlist and cracked modes +# --------------------------------------------------------------------------- + + +def test_sample_helper_is_module_level(): + """The sampling logic lives in one shared, module-level helper.""" + assert callable(hc_main._sample_plaintext_file) + + +def test_sample_helper_caps_and_labels_source(tmp_path, capsys): + src = tmp_path / "src.txt" + src.write_text("\n".join(f"p{i:04d}" for i in range(100)) + "\n") + + sampled = hc_main._sample_plaintext_file(str(src), 10, source_label="cracked passwords") + + assert sampled is not None + assert len(sampled) == 10 + captured = capsys.readouterr() + assert "Sampled 10 of 100 passwords from cracked passwords." in captured.out + + +def test_sample_helper_returns_none_on_read_error(tmp_path, capsys): + missing = tmp_path / "nope.txt" + with mock.patch("builtins.open", side_effect=OSError("boom")): + assert hc_main._sample_plaintext_file(str(missing), 10) is None + assert "Error reading wordlist: boom" in capsys.readouterr().out + + +def test_sample_helper_empty_file_returns_empty_list(tmp_path): + src = tmp_path / "empty.txt" + src.write_text("\n\n") + assert hc_main._sample_plaintext_file(str(src), 10) == [] + + +def test_cracked_mode_uses_shared_sampling_helper(env): + """cracked mode must route through _sample_plaintext_file, not its own copy.""" + out_path = env.hash_file + ".out" + with open(out_path, "w") as f: + f.write("hash:Summer2024!\n") + + with _ollama_globals(env.tmp_path, max_sample=123), mock.patch.object( + hc_main, "_sample_plaintext_file", return_value=["Summer2024!"] + ) as sampler, mock.patch.object( + hc_main.llm, "generate_candidates", return_value=["Winter2025!"] + ), mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("0", env.hash_file, "cracked", None) + + sampler.assert_called_once_with( + out_path, 123, source_label="cracked passwords" + ) + + +def test_wordlist_mode_uses_shared_sampling_helper(env): + wl = env.tmp_path / "wl.txt" + wl.write_text("alpha\n") + + with _ollama_globals(env.tmp_path, max_sample=77), mock.patch.object( + hc_main, "_sample_plaintext_file", return_value=["alpha"] + ) as sampler, mock.patch.object( + hc_main.llm, "generate_candidates", return_value=["x"] + ), mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("0", env.hash_file, "wordlist", str(wl)) + + sampler.assert_called_once_with(str(wl), 77) + + +def test_cracked_mode_caps_large_out_file(env, capsys): + """A large .out file gets the same evenly-spaced capping as a wordlist.""" + out_path = env.hash_file + ".out" + with open(out_path, "w") as f: + f.write("\n".join(f"h{i}:pw{i:06d}" for i in range(1000)) + "\n") + + with _ollama_globals(env.tmp_path, max_sample=25), mock.patch.object( + hc_main.llm, "generate_candidates", return_value=["x"] + ) as gen, mock.patch("subprocess.Popen", return_value=_make_proc()): + hc_main.hcatOllama("0", env.hash_file, "cracked", None) + + sample_lines = gen.call_args[0][4]["sample"].splitlines() + assert len(sample_lines) == 25 + indices = [int(w.replace("pw", "")) for w in sample_lines] + assert min(indices) < 100 + assert max(indices) >= 900 + assert "Sampled 25 of 1,000 passwords from cracked passwords." in capsys.readouterr().out From 07ca6ba7e34857f7e1fc82b528d9aef920e4787c Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Fri, 24 Jul 2026 18:31:04 -0400 Subject: [PATCH 07/11] fix(ui): route ollama/omen submenus through interactive_menu; re-prompt pickers on invalid input - ollama_attack: convert generation-mode menu to interactive_menu with dynamic items list (option 3 only present when cracked file exists); cancel via 99 or Escape; re-prompts in a while loop - omen_attack: convert existing-model menu to interactive_menu; cancel via 99 or Escape (replaces hard-coded "3. Cancel"); re-prompts in loop - _omen_pick_training_wordlist: replace abort-on-invalid with re-prompt loop; add explicit 'q' cancel key so users can exit without being trapped; callers already handle None correctly - _markov_pick_training_source: same re-prompt-loop + 'q' cancel fix; callers already handle None correctly - Update all affected tests to patch interactive_menu at hate_crack.attacks.interactive_menu (module-level import path) and drive follow-up prompts via builtins.input - Add new tests: arrow-menu reach, Escape/99 cancel, re-prompt loop coverage for both pickers, and conditional option-3 key presence Co-Authored-By: Claude --- hate_crack/attacks.py | 189 ++++++++++++++++-------------- tests/test_attacks_behavior.py | 208 +++++++++++++++++++++++++++++---- tests/test_omen_attack.py | 54 ++++++--- 3 files changed, 326 insertions(+), 125 deletions(-) diff --git a/hate_crack/attacks.py b/hate_crack/attacks.py index c6252e2..ac4560c 100644 --- a/hate_crack/attacks.py +++ b/hate_crack/attacks.py @@ -512,65 +512,73 @@ def bandrel_method(ctx: Any) -> None: def ollama_attack(ctx: Any) -> None: _notify.prompt_notify_for_attack("LLM") - print("\n\tLLM Attack") - print("\t1. Target info (company / industry / location)") - print("\t2. Wordlist (generate basewords from a sample wordlist)") # Cracked-password mode is only offered when this session actually has # plaintexts to learn from, matching _markov_pick_training_source. out_path = f"{ctx.hcatHashFile}.out" has_cracked = os.path.isfile(out_path) and os.path.getsize(out_path) > 0 - if has_cracked: - print("\t3. Cracked passwords (current session)") - choice = input("\n\tSelect generation mode: ").strip() - if choice == "1": - company = input("Company name: ").strip() - industry = input("Industry: ").strip() - location = input("Location: ").strip() - ctx.hcatOllama( - ctx.hcatHashType, - ctx.hcatHashFile, - "target", - {"company": company, "industry": industry, "location": location}, - ) - elif choice == "2": - path = _omen_pick_training_wordlist(ctx, title="LLM Sample Wordlists") - if not path: + items: list[tuple[str, str]] = [ + ("1", "Target info (company / industry / location)"), + ("2", "Wordlist (generate basewords from a sample wordlist)"), + ] + if has_cracked: + items.append(("3", "Cracked passwords (current session)")) + items.append(("99", "Cancel")) + + while True: + choice = interactive_menu(items, title="\nLLM Attack", prompt="\n\tSelect generation mode: ") + if choice is None or choice == "99": + return + if choice == "1": + company = input("Company name: ").strip() + industry = input("Industry: ").strip() + location = input("Location: ").strip() + ctx.hcatOllama( + ctx.hcatHashType, + ctx.hcatHashFile, + "target", + {"company": company, "industry": industry, "location": location}, + ) + return + elif choice == "2": + path = _omen_pick_training_wordlist(ctx, title="LLM Sample Wordlists") + if not path: + return + ctx.hcatOllama(ctx.hcatHashType, ctx.hcatHashFile, "wordlist", path) + return + elif choice == "3" and has_cracked: + ctx.hcatOllama(ctx.hcatHashType, ctx.hcatHashFile, "cracked", out_path) return - ctx.hcatOllama(ctx.hcatHashType, ctx.hcatHashFile, "wordlist", path) - elif choice == "3" and has_cracked: - ctx.hcatOllama(ctx.hcatHashType, ctx.hcatHashFile, "cracked", out_path) - elif choice == "3": - print("\t[!] No cracked passwords yet — crack some hashes first.") - else: - print("\t[!] Invalid selection.") def _omen_pick_training_wordlist(ctx: Any, title: str = "Training Wordlists"): - """Show wordlist picker. Returns path or None.""" + """Show wordlist picker. Returns path or None (user cancelled with 'q').""" wordlist_files = ctx.list_wordlist_files(ctx.hcatWordlists) - if wordlist_files: - entries = [f"{i}) {f}" for i, f in enumerate(wordlist_files, start=1)] - max_len = max((len(e) for e in entries), default=24) - print_multicolumn_list( - title, - entries, - min_col_width=max_len, - max_col_width=max_len, - ) - print("\tp. Enter a custom path") - sel = input("\n\tSelect wordlist: ").strip() - if sel.lower() == "p": - path = input("\n\tPath to wordlist: ").strip() - return path if path else None - try: - idx = int(sel) - if 1 <= idx <= len(wordlist_files): - return os.path.join(ctx.hcatWordlists, wordlist_files[idx - 1]) - except (ValueError, IndexError): - pass - print("\t[!] Invalid selection.") - return None + while True: + if wordlist_files: + entries = [f"{i}) {f}" for i, f in enumerate(wordlist_files, start=1)] + max_len = max((len(e) for e in entries), default=24) + print_multicolumn_list( + title, + entries, + min_col_width=max_len, + max_col_width=max_len, + ) + print("\tp. Enter a custom path") + print("\tq. Cancel") + sel = input("\n\tSelect wordlist: ").strip() + if sel.lower() == "q": + return None + if sel.lower() == "p": + path = input("\n\tPath to wordlist: ").strip() + return path if path else None + try: + idx = int(sel) + if 1 <= idx <= len(wordlist_files): + return os.path.join(ctx.hcatWordlists, wordlist_files[idx - 1]) + except (ValueError, IndexError): + pass + print("\t[!] Invalid selection.") def omen_attack(ctx: Any) -> None: @@ -592,16 +600,20 @@ def omen_attack(ctx: Any) -> None: info = ctx._omen_model_info(model_dir) trained_with = info.get("training_file", "unknown") if info else "unknown" print(f"\n\tOMEN model found (trained with: {trained_with})") - print("\t1. Use existing model") - print("\t2. Train new model (overwrites existing)") - print("\t3. Cancel") - choice = input("\n\tChoice: ").strip() - if choice == "1": - need_training = False - elif choice == "3": - return - elif choice != "2": - return + model_items = [ + ("1", "Use existing model"), + ("2", "Train new model (overwrites existing)"), + ("99", "Cancel"), + ] + while True: + choice = interactive_menu(model_items, title="\nOMEN Attack (Ordered Markov ENumerator)", prompt="\n\tChoice: ") + if choice is None or choice == "99": + return + if choice == "1": + need_training = False + break + elif choice == "2": + break else: print("\n\tNo valid OMEN model found. Training is required.") @@ -628,38 +640,41 @@ def omen_attack(ctx: Any) -> None: def _markov_pick_training_source(ctx: Any): - """Prompt user to select markov training source. Returns file path or None.""" + """Prompt user to select markov training source. Returns file path or None (user cancelled with 'q').""" out_path = f"{ctx.hcatHashFile}.out" has_cracked = os.path.isfile(out_path) and os.path.getsize(out_path) > 0 wordlist_files = ctx.list_wordlist_files(ctx.hcatWordlists) - entries = [] - if has_cracked: - entries.append("0) Cracked passwords (current session)") - entries.extend([f"{i}) {f}" for i, f in enumerate(wordlist_files, start=1)]) - if entries: - max_len = max((len(e) for e in entries), default=24) - print_multicolumn_list( - "Markov Training Source", - entries, - min_col_width=max_len, - max_col_width=max_len, - ) - print("\tp. Enter a custom path") - sel = input("\n\tSelect training source: ").strip() - if sel == "0" and has_cracked: - return out_path - if sel.lower() == "p": - path = input("\n\tPath to training file: ").strip() - return path if path else None - try: - idx = int(sel) - if 1 <= idx <= len(wordlist_files): - return os.path.join(ctx.hcatWordlists, wordlist_files[idx - 1]) - except (ValueError, IndexError): - pass - print("\t[!] Invalid selection.") - return None + while True: + entries = [] + if has_cracked: + entries.append("0) Cracked passwords (current session)") + entries.extend([f"{i}) {f}" for i, f in enumerate(wordlist_files, start=1)]) + if entries: + max_len = max((len(e) for e in entries), default=24) + print_multicolumn_list( + "Markov Training Source", + entries, + min_col_width=max_len, + max_col_width=max_len, + ) + print("\tp. Enter a custom path") + print("\tq. Cancel") + sel = input("\n\tSelect training source: ").strip() + if sel.lower() == "q": + return None + if sel == "0" and has_cracked: + return out_path + if sel.lower() == "p": + path = input("\n\tPath to training file: ").strip() + return path if path else None + try: + idx = int(sel) + if 1 <= idx <= len(wordlist_files): + return os.path.join(ctx.hcatWordlists, wordlist_files[idx - 1]) + except (ValueError, IndexError): + pass + print("\t[!] Invalid selection.") def adhoc_mask_crack(ctx: Any) -> None: diff --git a/tests/test_attacks_behavior.py b/tests/test_attacks_behavior.py index edee3aa..ae7186b 100644 --- a/tests/test_attacks_behavior.py +++ b/tests/test_attacks_behavior.py @@ -329,7 +329,10 @@ class TestOllamaAttack: def test_calls_hcatOllama_with_context(self) -> None: ctx = _make_ctx() - with patch("builtins.input", side_effect=["1", "ACME", "tech", "NYC"]): + with ( + patch("hate_crack.attacks.interactive_menu", return_value="1"), + patch("builtins.input", side_effect=["ACME", "tech", "NYC"]), + ): ollama_attack(ctx) ctx.hcatOllama.assert_called_once_with( @@ -342,7 +345,10 @@ class TestOllamaAttack: def test_passes_hash_type_and_file(self) -> None: ctx = _make_ctx(hash_type="1800", hash_file="/tmp/sha512.txt") - with patch("builtins.input", side_effect=["1", "Corp", "finance", "London"]): + with ( + patch("hate_crack.attacks.interactive_menu", return_value="1"), + patch("builtins.input", side_effect=["Corp", "finance", "London"]), + ): ollama_attack(ctx) call_args = ctx.hcatOllama.call_args[0] @@ -352,7 +358,10 @@ class TestOllamaAttack: def test_strips_whitespace_from_inputs(self) -> None: ctx = _make_ctx() - with patch("builtins.input", side_effect=["1", " ACME ", " tech ", " NYC "]): + with ( + patch("hate_crack.attacks.interactive_menu", return_value="1"), + patch("builtins.input", side_effect=[" ACME ", " tech ", " NYC "]), + ): ollama_attack(ctx) target_info = ctx.hcatOllama.call_args[0][3] @@ -363,7 +372,10 @@ class TestOllamaAttack: def test_target_string_is_literal_target(self) -> None: ctx = _make_ctx() - with patch("builtins.input", side_effect=["1", "X", "Y", "Z"]): + with ( + patch("hate_crack.attacks.interactive_menu", return_value="1"), + patch("builtins.input", side_effect=["X", "Y", "Z"]), + ): ollama_attack(ctx) assert ctx.hcatOllama.call_args[0][2] == "target" @@ -373,30 +385,44 @@ class TestOllamaAttack: ctx.list_wordlist_files.return_value = ["rockyou.txt"] ctx.hcatWordlists = "/tmp/wl" - # mode "2", then pick wordlist "1" - with patch("builtins.input", side_effect=["2", "1"]): + # mode "2" from interactive_menu, then pick wordlist "1" via input + with ( + patch("hate_crack.attacks.interactive_menu", return_value="2"), + patch("builtins.input", side_effect=["1"]), + ): ollama_attack(ctx) args = ctx.hcatOllama.call_args[0] assert args[2] == "wordlist" assert args[3].endswith("rockyou.txt") - def test_invalid_mode_does_not_call_hcatOllama(self) -> None: + def test_escape_cancels_without_calling_hcatOllama(self) -> None: + """None from interactive_menu (Escape / 99) cancels the attack.""" ctx = _make_ctx() - with patch("builtins.input", side_effect=["9"]): + with patch("hate_crack.attacks.interactive_menu", return_value=None): + ollama_attack(ctx) + ctx.hcatOllama.assert_not_called() + + def test_cancel_key_cancels_without_calling_hcatOllama(self) -> None: + """Selecting '99' (Cancel) cancels the attack.""" + ctx = _make_ctx() + with patch("hate_crack.attacks.interactive_menu", return_value="99"): ollama_attack(ctx) ctx.hcatOllama.assert_not_called() def test_wordlist_mode_aborts_when_no_wordlist_picked(self) -> None: ctx = _make_ctx() - # mode "2", then an invalid picker selection -> picker returns None + # mode "2" from interactive_menu, then user cancels the file picker ctx.list_wordlist_files.return_value = [] - with patch("builtins.input", side_effect=["2", "nonsense"]): + with ( + patch("hate_crack.attacks.interactive_menu", return_value="2"), + patch("builtins.input", side_effect=["q"]), + ): ollama_attack(ctx) ctx.hcatOllama.assert_not_called() def test_cracked_mode_offered_when_out_file_has_content( - self, tmp_path: Path, capsys + self, tmp_path: Path ) -> None: hash_file = tmp_path / "hashes.txt" hash_file.touch() @@ -404,43 +430,63 @@ class TestOllamaAttack: out_file.write_text("hash:Summer2024!\n") ctx = _make_ctx(hash_file=str(hash_file)) - with patch("builtins.input", side_effect=["3"]): + captured_items: list[list[tuple[str, str]]] = [] + + def capture_menu(items, **kwargs): + captured_items.append(list(items)) + return "3" + + with patch("hate_crack.attacks.interactive_menu", side_effect=capture_menu): ollama_attack(ctx) - assert "3. Cracked passwords" in capsys.readouterr().out + # Option 3 must be present in the items list when cracked file exists + keys = [k for k, _ in captured_items[0]] + assert "3" in keys ctx.hcatOllama.assert_called_once_with( ctx.hcatHashType, str(hash_file), "cracked", str(out_file) ) def test_cracked_mode_not_offered_when_out_file_missing( - self, tmp_path: Path, capsys + self, tmp_path: Path ) -> None: + """Option 3 must NOT appear in items when no cracked file exists.""" hash_file = tmp_path / "hashes.txt" hash_file.touch() ctx = _make_ctx(hash_file=str(hash_file)) - with patch("builtins.input", side_effect=["3"]): + captured_items: list[list[tuple[str, str]]] = [] + + def capture_menu(items, **kwargs): + captured_items.append(list(items)) + return "99" # cancel + + with patch("hate_crack.attacks.interactive_menu", side_effect=capture_menu): ollama_attack(ctx) - out = capsys.readouterr().out - assert "3. Cracked passwords" not in out - assert "No cracked passwords yet" in out + keys = [k for k, _ in captured_items[0]] + assert "3" not in keys ctx.hcatOllama.assert_not_called() def test_cracked_mode_not_offered_when_out_file_empty( - self, tmp_path: Path, capsys + self, tmp_path: Path ) -> None: + """Option 3 must NOT appear in items when cracked file exists but is empty.""" hash_file = tmp_path / "hashes.txt" hash_file.touch() (tmp_path / "hashes.txt.out").touch() # exists but zero bytes ctx = _make_ctx(hash_file=str(hash_file)) - with patch("builtins.input", side_effect=["3"]): + captured_items: list[list[tuple[str, str]]] = [] + + def capture_menu(items, **kwargs): + captured_items.append(list(items)) + return "99" # cancel + + with patch("hate_crack.attacks.interactive_menu", side_effect=capture_menu): ollama_attack(ctx) - out = capsys.readouterr().out - assert "3. Cracked passwords" not in out - assert "No cracked passwords yet" in out + keys = [k for k, _ in captured_items[0]] + assert "3" not in keys ctx.hcatOllama.assert_not_called() def test_target_and_wordlist_modes_unaffected_by_cracked_option( @@ -452,7 +498,121 @@ class TestOllamaAttack: (tmp_path / "hashes.txt.out").write_text("hash:Summer2024!\n") ctx = _make_ctx(hash_file=str(hash_file)) - with patch("builtins.input", side_effect=["1", "ACME", "tech", "NYC"]): + with ( + patch("hate_crack.attacks.interactive_menu", return_value="1"), + patch("builtins.input", side_effect=["ACME", "tech", "NYC"]), + ): ollama_attack(ctx) assert ctx.hcatOllama.call_args[0][2] == "target" + + def test_arrow_menu_env_reaches_ollama_attack(self) -> None: + """HATE_CRACK_ARROW_MENU=1 routes through interactive_menu in ollama_attack.""" + import os + from hate_crack.attacks import interactive_menu as real_im + + ctx = _make_ctx() + calls: list[tuple] = [] + + def spy_menu(items, **kwargs): + calls.append(tuple(items)) + return "99" # cancel immediately + + with ( + patch("hate_crack.attacks.interactive_menu", side_effect=spy_menu), + ): + ollama_attack(ctx) + + # interactive_menu was called — confirms the arrow-menu code path is reachable + assert len(calls) == 1 + keys = [k for k, _ in calls[0]] + assert "1" in keys + assert "2" in keys + assert "99" in keys + + +class TestOmenPickTrainingWordlistReprompt: + """_omen_pick_training_wordlist re-prompts on invalid input instead of aborting.""" + + def _make_ctx(self, wordlist_files=None): + ctx = MagicMock() + ctx.list_wordlist_files.return_value = wordlist_files or ["rockyou.txt"] + ctx.hcatWordlists = "/tmp/wl" + ctx.hcatHashFile = "/tmp/hashes.txt" + return ctx + + def test_invalid_input_reprompts_then_valid_pick(self) -> None: + from hate_crack.attacks import _omen_pick_training_wordlist + + ctx = self._make_ctx(["rockyou.txt"]) + # First input is invalid, second is valid + with patch("builtins.input", side_effect=["bad", "1"]): + result = _omen_pick_training_wordlist(ctx) + assert result is not None + assert "rockyou.txt" in result + + def test_cancel_with_q_returns_none(self) -> None: + from hate_crack.attacks import _omen_pick_training_wordlist + + ctx = self._make_ctx(["rockyou.txt"]) + with patch("builtins.input", return_value="q"): + result = _omen_pick_training_wordlist(ctx) + assert result is None + + def test_multiple_invalid_inputs_then_cancel(self) -> None: + from hate_crack.attacks import _omen_pick_training_wordlist + + ctx = self._make_ctx(["rockyou.txt"]) + with patch("builtins.input", side_effect=["99", "abc", "q"]): + result = _omen_pick_training_wordlist(ctx) + assert result is None + + +class TestMarkovPickTrainingSourceReprompt: + """_markov_pick_training_source re-prompts on invalid input instead of aborting.""" + + def _make_ctx(self, tmp_path, has_cracked=False, wordlist_files=None): + ctx = MagicMock() + hash_file = str(tmp_path / "hashes.txt") + ctx.hcatHashFile = hash_file + ctx.list_wordlist_files.return_value = wordlist_files or ["rockyou.txt"] + ctx.hcatWordlists = str(tmp_path / "wordlists") + if has_cracked: + (tmp_path / "hashes.txt.out").write_text("cracked_pw\n") + return ctx + + def test_invalid_input_reprompts_then_valid_pick(self, tmp_path: Path) -> None: + from hate_crack.attacks import _markov_pick_training_source + + ctx = self._make_ctx(tmp_path, wordlist_files=["rockyou.txt"]) + with patch("builtins.input", side_effect=["bad", "1"]): + result = _markov_pick_training_source(ctx) + assert result is not None + assert "rockyou.txt" in result + + def test_cancel_with_q_returns_none(self, tmp_path: Path) -> None: + from hate_crack.attacks import _markov_pick_training_source + + ctx = self._make_ctx(tmp_path) + with patch("builtins.input", return_value="q"): + result = _markov_pick_training_source(ctx) + assert result is None + + def test_multiple_invalid_then_cancel(self, tmp_path: Path) -> None: + from hate_crack.attacks import _markov_pick_training_source + + ctx = self._make_ctx(tmp_path, wordlist_files=["rockyou.txt"]) + with patch("builtins.input", side_effect=["99", "abc", "q"]): + result = _markov_pick_training_source(ctx) + assert result is None + + def test_caller_handles_none_correctly(self, tmp_path: Path) -> None: + """markov_brute_force returns early without crashing when picker returns None.""" + from hate_crack.attacks import markov_brute_force + + ctx = self._make_ctx(tmp_path, wordlist_files=["rockyou.txt"]) + # No .hcstat2 file → goes straight to picker; user cancels + with patch("builtins.input", return_value="q"): + markov_brute_force(ctx) + ctx.hcatMarkovTrain.assert_not_called() + ctx.hcatMarkovBruteForce.assert_not_called() diff --git a/tests/test_omen_attack.py b/tests/test_omen_attack.py index ef33507..cc61f9e 100644 --- a/tests/test_omen_attack.py +++ b/tests/test_omen_attack.py @@ -277,8 +277,10 @@ class TestOmenAttackHandler: def test_use_existing_model(self, tmp_path): ctx = self._make_ctx(tmp_path, model_valid=True) self._setup_rules_dir(tmp_path) - with patch("os.path.isfile", return_value=True), patch( - "builtins.input", side_effect=["1", "", "0"] + with ( + patch("os.path.isfile", return_value=True), + patch("hate_crack.attacks.interactive_menu", return_value="1"), + patch("builtins.input", side_effect=["", "0"]), ): from hate_crack.attacks import omen_attack @@ -289,8 +291,10 @@ class TestOmenAttackHandler: def test_train_new_model_with_wordlist_pick(self, tmp_path): ctx = self._make_ctx(tmp_path, model_valid=True) self._setup_rules_dir(tmp_path) - with patch("os.path.isfile", return_value=True), patch( - "builtins.input", side_effect=["2", "1", "", "0"] + with ( + patch("os.path.isfile", return_value=True), + patch("hate_crack.attacks.interactive_menu", return_value="2"), + patch("builtins.input", side_effect=["1", "", "0"]), ): from hate_crack.attacks import omen_attack @@ -302,8 +306,22 @@ class TestOmenAttackHandler: def test_cancel_aborts(self, tmp_path): ctx = self._make_ctx(tmp_path, model_valid=True) - with patch("os.path.isfile", return_value=True), patch( - "builtins.input", side_effect=["3"] + with ( + patch("os.path.isfile", return_value=True), + patch("hate_crack.attacks.interactive_menu", return_value="99"), + ): + from hate_crack.attacks import omen_attack + + omen_attack(ctx) + ctx.hcatOmenTrain.assert_not_called() + ctx.hcatOmen.assert_not_called() + + def test_escape_cancels(self, tmp_path): + """None from interactive_menu (Escape) cancels the attack.""" + ctx = self._make_ctx(tmp_path, model_valid=True) + with ( + patch("os.path.isfile", return_value=True), + patch("hate_crack.attacks.interactive_menu", return_value=None), ): from hate_crack.attacks import omen_attack @@ -348,8 +366,10 @@ class TestOmenAttackHandler: def test_rules_passed_to_hcatOmen(self, tmp_path): ctx = self._make_ctx(tmp_path, model_valid=True) self._setup_rules_dir(tmp_path, ["best64.rule"]) - with patch("os.path.isfile", return_value=True), patch( - "builtins.input", side_effect=["1", "", "1"] + with ( + patch("os.path.isfile", return_value=True), + patch("hate_crack.attacks.interactive_menu", return_value="1"), + patch("builtins.input", side_effect=["", "1"]), ): from hate_crack.attacks import omen_attack @@ -361,8 +381,10 @@ class TestOmenAttackHandler: def test_multiple_rule_chains_spawn_multiple_calls(self, tmp_path): ctx = self._make_ctx(tmp_path, model_valid=True) self._setup_rules_dir(tmp_path, ["best64.rule", "dive.rule"]) - with patch("os.path.isfile", return_value=True), patch( - "builtins.input", side_effect=["1", "", "1,2"] + with ( + patch("os.path.isfile", return_value=True), + patch("hate_crack.attacks.interactive_menu", return_value="1"), + patch("builtins.input", side_effect=["", "1,2"]), ): from hate_crack.attacks import omen_attack @@ -372,8 +394,10 @@ class TestOmenAttackHandler: def test_cancel_from_rules_aborts(self, tmp_path): ctx = self._make_ctx(tmp_path, model_valid=True) self._setup_rules_dir(tmp_path, ["best64.rule"]) - with patch("os.path.isfile", return_value=True), patch( - "builtins.input", side_effect=["1", "", "99"] + with ( + patch("os.path.isfile", return_value=True), + patch("hate_crack.attacks.interactive_menu", return_value="1"), + patch("builtins.input", side_effect=["", "99"]), ): from hate_crack.attacks import omen_attack @@ -383,8 +407,10 @@ class TestOmenAttackHandler: def test_no_rules_passes_empty_chain(self, tmp_path): ctx = self._make_ctx(tmp_path, model_valid=True) self._setup_rules_dir(tmp_path, ["best64.rule"]) - with patch("os.path.isfile", return_value=True), patch( - "builtins.input", side_effect=["1", "", "0"] + with ( + patch("os.path.isfile", return_value=True), + patch("hate_crack.attacks.interactive_menu", return_value="1"), + patch("builtins.input", side_effect=["", "0"]), ): from hate_crack.attacks import omen_attack From 24e574bc8e487c28c6f30c95ebd06b0692aaf576 Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Fri, 24 Jul 2026 18:33:11 -0400 Subject: [PATCH 08/11] fix(ui): print picker grids once and report rejected menu keys The re-prompt loops repainted the whole multicolumn wordlist grid after every typo, burying the error message under dozens of entries. Hoist the grid and the p/q legend out of the loop so only the prompt repeats. ollama_attack's loop also fell through to a silent redraw on an unrecognised key, leaving no way to tell a rejected key from a repainted prompt. Co-Authored-By: Claude --- hate_crack/attacks.py | 58 ++++++++++++++++++++++++------------------- 1 file changed, 33 insertions(+), 25 deletions(-) diff --git a/hate_crack/attacks.py b/hate_crack/attacks.py index ac4560c..4288d94 100644 --- a/hate_crack/attacks.py +++ b/hate_crack/attacks.py @@ -549,23 +549,30 @@ def ollama_attack(ctx: Any) -> None: elif choice == "3" and has_cracked: ctx.hcatOllama(ctx.hcatHashType, ctx.hcatHashFile, "cracked", out_path) return + else: + # Without this the menu just silently redraws and the user cannot + # tell a rejected key from a repainted prompt. + print("\t[!] Invalid selection.") def _omen_pick_training_wordlist(ctx: Any, title: str = "Training Wordlists"): """Show wordlist picker. Returns path or None (user cancelled with 'q').""" wordlist_files = ctx.list_wordlist_files(ctx.hcatWordlists) + # Print the grid once, outside the retry loop: a wordlists directory can + # hold dozens of entries, and repainting the whole thing after every typo + # buries the error message. + if wordlist_files: + entries = [f"{i}) {f}" for i, f in enumerate(wordlist_files, start=1)] + max_len = max((len(e) for e in entries), default=24) + print_multicolumn_list( + title, + entries, + min_col_width=max_len, + max_col_width=max_len, + ) + print("\tp. Enter a custom path") + print("\tq. Cancel") while True: - if wordlist_files: - entries = [f"{i}) {f}" for i, f in enumerate(wordlist_files, start=1)] - max_len = max((len(e) for e in entries), default=24) - print_multicolumn_list( - title, - entries, - min_col_width=max_len, - max_col_width=max_len, - ) - print("\tp. Enter a custom path") - print("\tq. Cancel") sel = input("\n\tSelect wordlist: ").strip() if sel.lower() == "q": return None @@ -645,21 +652,22 @@ def _markov_pick_training_source(ctx: Any): has_cracked = os.path.isfile(out_path) and os.path.getsize(out_path) > 0 wordlist_files = ctx.list_wordlist_files(ctx.hcatWordlists) + # Print the grid once, outside the retry loop — see _omen_pick_training_wordlist. + entries = [] + if has_cracked: + entries.append("0) Cracked passwords (current session)") + entries.extend([f"{i}) {f}" for i, f in enumerate(wordlist_files, start=1)]) + if entries: + max_len = max((len(e) for e in entries), default=24) + print_multicolumn_list( + "Markov Training Source", + entries, + min_col_width=max_len, + max_col_width=max_len, + ) + print("\tp. Enter a custom path") + print("\tq. Cancel") while True: - entries = [] - if has_cracked: - entries.append("0) Cracked passwords (current session)") - entries.extend([f"{i}) {f}" for i, f in enumerate(wordlist_files, start=1)]) - if entries: - max_len = max((len(e) for e in entries), default=24) - print_multicolumn_list( - "Markov Training Source", - entries, - min_col_width=max_len, - max_col_width=max_len, - ) - print("\tp. Enter a custom path") - print("\tq. Cancel") sel = input("\n\tSelect training source: ").strip() if sel.lower() == "q": return None From 6448f550eee8cd816dca9c1ab997c363021771cb Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Fri, 24 Jul 2026 18:42:50 -0400 Subject: [PATCH 09/11] feat(llm): pre-fill LLM target industry/location from local model research Target-info mode (menu 12) asked for company, industry, and location as three blank prompts. Once the company name is known the local Ollama model can often supply the other two, so ask it and offer the answers as editable defaults. - llm.research_target(): TargetResearchInput/TargetResearchOutput schemas plus a research SystemPromptGenerator that tells the model to return empty strings when it does not genuinely recognize the organization, so an unknown small client yields blank prompts instead of a confident hallucination. APITimeoutError is translated to LLMTimeoutError like generate_candidates. - clean_research_field(): strips, collapses whitespace, caps at 80 chars, and turns anything non-string into "" so model output cannot be pasted unbounded into a prompt default. - main.hcatOllamaResearchTarget(): runs the call inside the existing spinner and degrades to blank suggestions on timeout or any other failure, so research can never block the attack. - attacks.ollama_attack(): shows suggestions as "Industry (freight rail): " defaults, labelled explicitly as the model's GUESSES rather than OSINT. - New ollamaAutoResearch config toggle (default true) in both config examples. Research uses only the configured local Ollama server; the client name is never sent to a search engine or third-party company-data API. Co-Authored-By: Claude --- README.md | 22 ++ config.json.example | 1 + hate_crack/attacks.py | 47 ++++- hate_crack/config.json.example | 1 + hate_crack/llm.py | 129 +++++++++++- hate_crack/main.py | 38 ++++ tests/test_llm.py | 110 ++++++++++ tests/test_ollama_target_research.py | 301 +++++++++++++++++++++++++++ 8 files changed, 642 insertions(+), 7 deletions(-) create mode 100644 tests/test_ollama_target_research.py diff --git a/README.md b/README.md index 6b10e8c..32e2738 100644 --- a/README.md +++ b/README.md @@ -473,11 +473,33 @@ The LLM Attack (option 12) uses Ollama to generate password candidates. Configur - **`ollamaNumCtx`** — Context window size for the model (default: `2048`). - **`ollamaTimeout`** — Seconds to wait for a generation response before giving up (default: `300`). Raise this if a large model is still loading into VRAM on the first request, which can otherwise exceed the timeout; hate_crack prints the elapsed timeout and this setting's name when it fires. - **`ollamaMaxSampleLines`** — Maximum number of lines drawn from the source file and included in the LLM prompt when using **Wordlist** or **Cracked passwords** mode (default: `500`). Lines are sampled evenly across the whole file so the prompt reflects the wordlist's full character range rather than just the head. Set to a larger value if the model has a big context window (`ollamaNumCtx`) and you want richer coverage; set it lower to reduce prompt size and generation latency. Values ≤ 0 are treated as 500. +- **`ollamaAutoResearch`** — When `true` (default), **Target info** mode asks the local model to suggest the industry and location as soon as you have typed the company name, and offers them as editable prompt defaults. Set to `false` to always get blank prompts (useful with a slow model, since research costs one extra round-trip before the attack starts). - The Ollama URL defaults to `http://localhost:11434` (override via the `OLLAMA_HOST` env var). Ensure Ollama is running and the model is pulled (`ollama pull qwen2.5:32b`) before using the LLM Attack — hate_crack no longer auto-pulls missing models. The attack offers three generation modes: 1. **Target info** — company / industry / location; the model derives candidates from those details. + + After you type the company name, hate_crack asks the same local model what it already knows about that organization and pre-fills the **Industry** and **Location** prompts with the answers, shown in parentheses: + + ``` + Company name: Acme Rail Services + + [!] The values in parentheses below are the local model's GUESSES, not verified OSINT. + Press Enter to accept, or type your own value to override. + Industry (freight rail maintenance): + Location (Omaha, Nebraska): + ``` + + Press Enter to accept a suggestion or type over it. These values are the model's recollection, **not OSINT** — treat them as a starting point, not intelligence about the client. The lookup uses only the local Ollama server, so the client name never leaves the host; there are no web or third-party API calls. If the model does not recognize the organization (the common case for small clients), it returns nothing and you get plain blank prompts: + + ``` + Company name: Acme Rail Services + Industry: + Location: + ``` + + A research failure — timeout, Ollama not running, empty answer — never blocks the attack; it just falls back to blank prompts. Set `ollamaAutoResearch` to `false` to skip research entirely. 2. **Wordlist** — derive basewords from a sample wordlist. 3. **Cracked passwords** — feed the plaintexts already recovered this session (`.out`) back to the model so it can infer the target organization's own password conventions (basewords, seasons, years, suffixes, leetspeak) and generate *new* candidates in the same style. This option is only listed once at least one hash has been cracked; the sample is capped by `ollamaMaxSampleLines` exactly like Wordlist mode. diff --git a/config.json.example b/config.json.example index 86bd196..ec89e9a 100644 --- a/config.json.example +++ b/config.json.example @@ -29,6 +29,7 @@ "ollamaNumCtx": 2048, "ollamaTimeout": 300, "ollamaMaxSampleLines": 500, + "ollamaAutoResearch": true, "omenTrainingList": "rockyou.txt", "omenMaxCandidates": 50000000, "pcfgRuleset": "DEFAULT", diff --git a/hate_crack/attacks.py b/hate_crack/attacks.py index 4288d94..c84ccf2 100644 --- a/hate_crack/attacks.py +++ b/hate_crack/attacks.py @@ -7,6 +7,7 @@ from typing import Any from hate_crack import notify as _notify from hate_crack.api import download_hashmob_rules from hate_crack.formatting import print_multicolumn_list +from hate_crack.llm import clean_research_field from hate_crack.menu import interactive_menu @@ -510,6 +511,47 @@ def bandrel_method(ctx: Any) -> None: ctx.hcatBandrel(ctx.hcatHashType, ctx.hcatHashFile) +def _research_target_suggestions(ctx: Any, company: str) -> dict[str, str]: + """Ask the local model for industry/location suggestions for *company*. + + Returns a dict of cleaned suggestion strings (values may be ''). Research is + a convenience only, so any failure is swallowed here as well as in + ``hcatOllamaResearchTarget``: the operator still gets blank prompts and the + attack proceeds. + """ + if not company: + return {} + + try: + raw = ctx.hcatOllamaResearchTarget(company) + except Exception as e: + print(f"Note: target research unavailable ({e}) — enter the details manually.") + return {} + + suggestions = {} + if isinstance(raw, dict): + for key in ("industry", "location"): + value = clean_research_field(raw.get(key, "")) + if value: + suggestions[key] = value + + if suggestions: + print( + "\n[!] The values in parentheses below are the local model's GUESSES, " + "not verified OSINT." + ) + print(" Press Enter to accept, or type your own value to override.") + return suggestions + + +def _prompt_with_default(label: str, default: Any) -> str: + """Prompt for *label*, showing *default* in parentheses when there is one.""" + suggestion = clean_research_field(default) + if suggestion: + return input(f"{label} ({suggestion}): ").strip() or suggestion + return input(f"{label}: ").strip() + + def ollama_attack(ctx: Any) -> None: _notify.prompt_notify_for_attack("LLM") # Cracked-password mode is only offered when this session actually has @@ -531,8 +573,9 @@ def ollama_attack(ctx: Any) -> None: return if choice == "1": company = input("Company name: ").strip() - industry = input("Industry: ").strip() - location = input("Location: ").strip() + suggestions = _research_target_suggestions(ctx, company) + industry = _prompt_with_default("Industry", suggestions.get("industry")) + location = _prompt_with_default("Location", suggestions.get("location")) ctx.hcatOllama( ctx.hcatHashType, ctx.hcatHashFile, diff --git a/hate_crack/config.json.example b/hate_crack/config.json.example index 0bd02d7..fe8ffb8 100644 --- a/hate_crack/config.json.example +++ b/hate_crack/config.json.example @@ -26,6 +26,7 @@ "ollamaNumCtx": 2048, "ollamaTimeout": 300, "ollamaMaxSampleLines": 500, + "ollamaAutoResearch": true, "passgptModel": "javirandor/passgpt-10characters", "passgptMaxCandidates": 1000000, "passgptBatchSize": 1024, diff --git a/hate_crack/llm.py b/hate_crack/llm.py index f4cf5f5..9600489 100644 --- a/hate_crack/llm.py +++ b/hate_crack/llm.py @@ -1,7 +1,7 @@ """Structured LLM password-candidate generation via Atomic Agents + Ollama. Isolates the atomic-agents / instructor dependency. The rest of hate_crack talks -to this module only through ``generate_candidates``. +to this module only through ``generate_candidates`` and ``research_target``. """ import instructor @@ -14,6 +14,10 @@ from atomic_agents.context import SystemPromptGenerator MAX_CANDIDATE_LEN = 128 DEFAULT_TIMEOUT_SECONDS = 300.0 +# Researched target fields are pasted into an interactive prompt as an editable +# default, so they must stay short enough to fit on one terminal line. +MAX_RESEARCH_FIELD_LEN = 80 + class LLMTimeoutError(Exception): """The LLM server accepted the request but did not respond in time. @@ -43,6 +47,60 @@ class PasswordCandidatesOutput(BaseIOSchema): ) +class TargetResearchInput(BaseIOSchema): + """The company name to recall industry and location details for.""" + + company: str = Field( + ..., description="The name of the target organization, exactly as typed." + ) + + +class TargetResearchOutput(BaseIOSchema): + """Recalled industry and location for a named organization, or empty strings.""" + + industry: str = Field( + ..., + description=( + "The organization's industry or sector as a short phrase, e.g. " + "'regional healthcare provider' or 'commercial construction'. Return " + "an empty string if you do not actually recognize this organization." + ), + ) + location: str = Field( + ..., + description=( + "The organization's primary location as 'City, State/Country'. Return " + "an empty string if you do not actually recognize this organization." + ), + ) + + +_RESEARCH_PROMPT = SystemPromptGenerator( + background=[ + "You are assisting a security professional on an authorized penetration " + "test who is about to generate password candidates for a named client " + "organization.", + "You have no internet access. You may only answer from what you already " + "know about the organization.", + "Most client organizations are small and will be completely unknown to " + "you. That is the expected case, not a failure.", + ], + steps=[ + "Decide whether you genuinely recognize this specific organization by name.", + "If you do, recall its industry or sector and its primary location.", + "If you do not recognize it, or you are not reasonably confident, do not " + "guess and do not infer anything from the words in the name.", + ], + output_instructions=[ + "Return an empty string for any field you are not reasonably confident " + "about. An empty field is correct and useful; a fabricated one is harmful " + "because the operator may mistake it for real intelligence.", + "Keep each field under 80 characters.", + "Return only the industry and location fields — no explanations, " + "hedging, caveats, or commentary.", + ], +) + _TARGET_PROMPT = SystemPromptGenerator( background=[ "You are a security professional generating password candidates during an " @@ -135,6 +193,70 @@ def _build_request(mode: str, context_data: dict) -> str: raise ValueError(f"Unknown LLM generation mode: {mode}") +def _build_client(url: str, timeout: float) -> instructor.Instructor: + """Build the instructor-wrapped OpenAI client pointed at an Ollama server.""" + return instructor.from_openai( + OpenAI(base_url=f"{url}/v1", api_key="ollama", timeout=timeout), + mode=instructor.Mode.JSON, + ) + + +def clean_research_field(value: object) -> str: + """Strip and length-cap one researched field; return '' for no suggestion. + + Anything that is not a non-empty string after stripping — including a model + that echoed whitespace or an over-long ramble — becomes '' so the caller + falls back to a plain blank prompt instead of pasting model noise into it. + """ + if not isinstance(value, str): + return "" + cleaned = " ".join(value.split()) + if len(cleaned) > MAX_RESEARCH_FIELD_LEN: + cleaned = cleaned[:MAX_RESEARCH_FIELD_LEN].rstrip() + return cleaned + + +def research_target( + url: str, + model: str, + num_ctx: int, + company: str, + timeout: float = DEFAULT_TIMEOUT_SECONDS, +) -> TargetResearchOutput: + """Ask the local model what it already knows about *company*. + + Returns a ``TargetResearchOutput`` whose ``industry`` and ``location`` are + stripped and capped at ``MAX_RESEARCH_FIELD_LEN``; either may be '' when the + model is not confident, which callers must treat as "no suggestion". + + Uses only the configured local Ollama server — no web lookups, so the client + name never leaves the host. Raises LLMTimeoutError if the request exceeds + ``timeout``; other client/connection errors propagate to the caller. + """ + client = _build_client(url, timeout) + + agent = AtomicAgent[TargetResearchInput, TargetResearchOutput]( + config=AgentConfig( + client=client, + model=model, + system_prompt_generator=_RESEARCH_PROMPT, + model_api_parameters={"extra_body": {"options": {"num_ctx": num_ctx}}}, + ) + ) + + try: + result = agent.run(TargetResearchInput(company=company)) + except APITimeoutError as e: + raise LLMTimeoutError( + f"no response from {url} within {timeout:g} seconds" + ) from e + + return TargetResearchOutput( + industry=clean_research_field(getattr(result, "industry", "")), + location=clean_research_field(getattr(result, "location", "")), + ) + + def generate_candidates( url: str, model: str, @@ -156,10 +278,7 @@ def generate_candidates( """ request = _build_request(mode, context_data) - client = instructor.from_openai( - OpenAI(base_url=f"{url}/v1", api_key="ollama", timeout=timeout), - mode=instructor.Mode.JSON, - ) + client = _build_client(url, timeout) # _build_request has already rejected unknown modes, so this lookup is safe. prompt_generator = _PROMPTS[mode] diff --git a/hate_crack/main.py b/hate_crack/main.py index b0818eb..f6dafd0 100755 --- a/hate_crack/main.py +++ b/hate_crack/main.py @@ -455,6 +455,7 @@ ollamaModel = config_parser.get("ollamaModel", "qwen2.5:32b") ollamaNumCtx = int(config_parser.get("ollamaNumCtx", 2048)) ollamaTimeout = float(config_parser.get("ollamaTimeout", 300)) ollamaMaxSampleLines = int(config_parser.get("ollamaMaxSampleLines", 500)) +ollamaAutoResearch = bool(config_parser.get("ollamaAutoResearch", True)) omenTrainingList = config_parser.get("omenTrainingList", "rockyou.txt") omenMaxCandidates = int(config_parser.get("omenMaxCandidates", 1000000)) @@ -2123,6 +2124,43 @@ def _sample_plaintext_file(path, cap, source_label="wordlist"): return sampled +def hcatOllamaResearchTarget(company): + """Ask the local Ollama model what it knows about *company*. + + Returns a dict with "industry" and "location" keys; either value may be an + empty string when the model is not confident or the request failed. Never + raises: research is a convenience, so any failure degrades to empty + suggestions (blank prompts) rather than blocking the attack. + + Uses only the configured local Ollama server — the company name is never + sent to a third-party service. + """ + blank = {"industry": "", "location": ""} + if not ollamaAutoResearch or not company: + return blank + + try: + with spinner(f"Researching {company} via Ollama ({ollamaModel})..."): + result = llm.research_target( + ollamaUrl, + ollamaModel, + ollamaNumCtx, + company, + timeout=ollamaTimeout, + ) + except llm.LLMTimeoutError: + print( + f"Note: target research timed out after {ollamaTimeout:g} seconds — " + "enter the details manually." + ) + return blank + except Exception as e: + print(f"Note: target research unavailable ({e}) — enter the details manually.") + return blank + + return {"industry": result.industry, "location": result.location} + + # LLM Ollama Attack def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): candidates_path = f"{hcatHashFile}.ollama_candidates" diff --git a/tests/test_llm.py b/tests/test_llm.py index 70c58ed..1e8ef94 100644 --- a/tests/test_llm.py +++ b/tests/test_llm.py @@ -210,3 +210,113 @@ def test_api_timeout_reraised_as_domain_error(): "target", {"company": "X", "industry": "Y", "location": "Z"}, timeout=1.0, ) + + +# --------------------------------------------------------------------------- +# research_target +# --------------------------------------------------------------------------- + + +def _patch_research_agent(industry, location): + """Patch client builders + AtomicAgent for a research call. No network.""" + result = mock.MagicMock() + result.industry = industry + result.location = location + + agent_instance = mock.MagicMock() + agent_instance.run.return_value = result + + agent_cls = mock.MagicMock() + agent_cls.__getitem__.return_value.return_value = agent_instance + + return ( + mock.patch( + "hate_crack.llm.instructor.from_openai", + return_value=mock.MagicMock(spec=instructor.Instructor), + ), + mock.patch("hate_crack.llm.OpenAI"), + mock.patch("hate_crack.llm.AtomicAgent", agent_cls), + agent_cls, + agent_instance, + ) + + +def test_research_target_returns_fields_and_passes_company(): + p_instr, p_openai, p_agent, agent_cls, agent_instance = _patch_research_agent( + "freight rail maintenance", "Omaha, Nebraska" + ) + with p_instr, p_openai, p_agent: + out = llm.research_target( + "http://localhost:11434", "qwen2.5:32b", 2048, "Acme Rail Services" + ) + assert out.industry == "freight rail maintenance" + assert out.location == "Omaha, Nebraska" + assert agent_instance.run.call_args[0][0].company == "Acme Rail Services" + + +def test_research_target_uses_research_prompt_and_num_ctx(): + p_instr, p_openai, p_agent, agent_cls, agent_instance = _patch_research_agent( + "x", "y" + ) + with p_instr, p_openai, p_agent: + llm.research_target("http://localhost:11434", "qwen2.5:32b", 4096, "Acme") + config = agent_cls.__getitem__.return_value.call_args.kwargs["config"] + assert config.system_prompt_generator is llm._RESEARCH_PROMPT + assert config.model_api_parameters["extra_body"]["options"]["num_ctx"] == 4096 + + +def test_research_prompt_tells_model_to_return_empty_when_unsure(): + rendered = llm._RESEARCH_PROMPT.generate_prompt().lower() + assert "empty string" in rendered + assert "no internet access" in rendered + + +def test_research_target_strips_and_blanks_whitespace_only(): + p_instr, p_openai, p_agent, agent_cls, agent_instance = _patch_research_agent( + " healthcare ", " " + ) + with p_instr, p_openai, p_agent: + out = llm.research_target("http://localhost:11434", "m", 2048, "Acme") + assert out.industry == "healthcare" + assert out.location == "" + + +def test_research_target_caps_overlong_values(): + p_instr, p_openai, p_agent, agent_cls, agent_instance = _patch_research_agent( + "A" * 500, "B" * (llm.MAX_RESEARCH_FIELD_LEN + 1) + ) + with p_instr, p_openai, p_agent: + out = llm.research_target("http://localhost:11434", "m", 2048, "Acme") + assert len(out.industry) == llm.MAX_RESEARCH_FIELD_LEN + assert len(out.location) == llm.MAX_RESEARCH_FIELD_LEN + + +def test_research_target_tolerates_non_string_fields(): + p_instr, p_openai, p_agent, agent_cls, agent_instance = _patch_research_agent( + None, 42 + ) + with p_instr, p_openai, p_agent: + out = llm.research_target("http://localhost:11434", "m", 2048, "Acme") + assert out.industry == "" + assert out.location == "" + + +def test_research_target_timeout_forwarded_and_translated(): + p_instr, p_openai, p_agent, agent_cls, agent_instance = _patch_research_agent( + "x", "y" + ) + agent_instance.run.side_effect = openai.APITimeoutError( + request=httpx.Request("POST", "http://localhost:11434/v1/chat/completions") + ) + with p_instr, p_openai as openai_cls, p_agent: + with pytest.raises(llm.LLMTimeoutError): + llm.research_target( + "http://localhost:11434", "m", 2048, "Acme", timeout=7.5 + ) + assert openai_cls.call_args.kwargs["timeout"] == 7.5 + + +def test_clean_research_field_collapses_internal_whitespace(): + assert llm.clean_research_field("commercial \n construction") == ( + "commercial construction" + ) diff --git a/tests/test_ollama_target_research.py b/tests/test_ollama_target_research.py new file mode 100644 index 0000000..222240a --- /dev/null +++ b/tests/test_ollama_target_research.py @@ -0,0 +1,301 @@ +"""Tests for LLM target-research pre-fill (menu 12, Target info mode). + +Covers both layers: hcatOllamaResearchTarget in main (spinner + failure +degradation) and ollama_attack's editable-default prompts in attacks. The LLM is +always mocked; no network. +""" + +import os +from unittest import mock +from unittest.mock import MagicMock, patch + +os.environ["HATE_CRACK_SKIP_INIT"] = "1" +from hate_crack import llm # noqa: E402 +from hate_crack import main as hc_main # noqa: E402 +from hate_crack.attacks import ollama_attack # noqa: E402 + + +def _make_ctx(hash_type: str = "1000", hash_file: str = "/tmp/hashes.txt") -> MagicMock: + ctx = MagicMock() + ctx.hcatHashType = hash_type + ctx.hcatHashFile = hash_file + return ctx + + +class _InputRecorder: + """input() stub that records prompts and replays canned answers.""" + + def __init__(self, answers): + self.answers = list(answers) + self.prompts: list[str] = [] + + def __call__(self, prompt=""): + self.prompts.append(prompt) + return self.answers.pop(0) + + +def _run_target_mode(ctx, answers): + recorder = _InputRecorder(answers) + with ( + patch("hate_crack.attacks.interactive_menu", return_value="1"), + patch("builtins.input", recorder), + ): + ollama_attack(ctx) + return recorder + + +# --------------------------------------------------------------------------- +# attacks layer: prompt defaults +# --------------------------------------------------------------------------- + + +class TestResearchPrefill: + def test_successful_research_prefills_defaults(self) -> None: + ctx = _make_ctx() + ctx.hcatOllamaResearchTarget.return_value = { + "industry": "freight rail maintenance", + "location": "Omaha, Nebraska", + } + + recorder = _run_target_mode(ctx, ["Acme Rail", "", ""]) + + ctx.hcatOllamaResearchTarget.assert_called_once_with("Acme Rail") + assert recorder.prompts == [ + "Company name: ", + "Industry (freight rail maintenance): ", + "Location (Omaha, Nebraska): ", + ] + + def test_enter_accepts_the_suggested_defaults(self) -> None: + ctx = _make_ctx() + ctx.hcatOllamaResearchTarget.return_value = { + "industry": "healthcare", + "location": "Austin, Texas", + } + + _run_target_mode(ctx, ["Acme", "", ""]) + + assert ctx.hcatOllama.call_args[0][3] == { + "company": "Acme", + "industry": "healthcare", + "location": "Austin, Texas", + } + + def test_typed_input_overrides_the_default(self) -> None: + ctx = _make_ctx() + ctx.hcatOllamaResearchTarget.return_value = { + "industry": "healthcare", + "location": "Austin, Texas", + } + + _run_target_mode(ctx, ["Acme", " banking ", "Berlin"]) + + assert ctx.hcatOllama.call_args[0][3] == { + "company": "Acme", + "industry": "banking", + "location": "Berlin", + } + + def test_partial_research_prefills_only_known_field(self) -> None: + ctx = _make_ctx() + ctx.hcatOllamaResearchTarget.return_value = { + "industry": "mining", + "location": "", + } + + recorder = _run_target_mode(ctx, ["Acme", "", "Perth"]) + + assert recorder.prompts[1] == "Industry (mining): " + assert recorder.prompts[2] == "Location: " + assert ctx.hcatOllama.call_args[0][3]["location"] == "Perth" + + def test_empty_research_falls_back_to_blank_prompts(self) -> None: + ctx = _make_ctx() + ctx.hcatOllamaResearchTarget.return_value = {"industry": "", "location": ""} + + recorder = _run_target_mode(ctx, ["Acme", "tech", "NYC"]) + + assert recorder.prompts == ["Company name: ", "Industry: ", "Location: "] + assert ctx.hcatOllama.call_args[0][3] == { + "company": "Acme", + "industry": "tech", + "location": "NYC", + } + + def test_whitespace_only_research_is_treated_as_no_suggestion(self) -> None: + ctx = _make_ctx() + ctx.hcatOllamaResearchTarget.return_value = { + "industry": " ", + "location": "\t\n", + } + + recorder = _run_target_mode(ctx, ["Acme", "", ""]) + + assert recorder.prompts == ["Company name: ", "Industry: ", "Location: "] + + def test_overlong_suggestion_is_capped_in_the_prompt(self) -> None: + ctx = _make_ctx() + ctx.hcatOllamaResearchTarget.return_value = { + "industry": "z" * 500, + "location": "", + } + + recorder = _run_target_mode(ctx, ["Acme", "", ""]) + + assert recorder.prompts[1] == f"Industry ({'z' * llm.MAX_RESEARCH_FIELD_LEN}): " + industry = ctx.hcatOllama.call_args[0][3]["industry"] + assert len(industry) == llm.MAX_RESEARCH_FIELD_LEN + + def test_research_exception_falls_back_and_does_not_abort(self) -> None: + """Even an unexpected raise from the research call must not block.""" + ctx = _make_ctx() + ctx.hcatOllamaResearchTarget.side_effect = RuntimeError("boom") + + recorder = _run_target_mode(ctx, ["Acme", "tech", "NYC"]) + + assert recorder.prompts == ["Company name: ", "Industry: ", "Location: "] + ctx.hcatOllama.assert_called_once() + + def test_non_dict_research_result_falls_back(self) -> None: + ctx = _make_ctx() + ctx.hcatOllamaResearchTarget.return_value = None + + recorder = _run_target_mode(ctx, ["Acme", "tech", "NYC"]) + + assert recorder.prompts == ["Company name: ", "Industry: ", "Location: "] + + def test_blank_company_skips_research_entirely(self) -> None: + ctx = _make_ctx() + + recorder = _run_target_mode(ctx, ["", "tech", "NYC"]) + + ctx.hcatOllamaResearchTarget.assert_not_called() + assert recorder.prompts == ["Company name: ", "Industry: ", "Location: "] + + def test_suggestions_are_labelled_as_guesses(self, capsys) -> None: + ctx = _make_ctx() + ctx.hcatOllamaResearchTarget.return_value = { + "industry": "healthcare", + "location": "Austin, Texas", + } + + _run_target_mode(ctx, ["Acme", "", ""]) + + out = capsys.readouterr().out + assert "GUESSES" in out + assert "not verified OSINT" in out + + def test_no_guess_banner_when_research_returns_nothing(self, capsys) -> None: + ctx = _make_ctx() + ctx.hcatOllamaResearchTarget.return_value = {"industry": "", "location": ""} + + _run_target_mode(ctx, ["Acme", "", ""]) + + assert "GUESSES" not in capsys.readouterr().out + + def test_hcatOllama_call_shape_unchanged(self) -> None: + ctx = _make_ctx(hash_type="1800", hash_file="/tmp/sha512.txt") + ctx.hcatOllamaResearchTarget.return_value = { + "industry": "healthcare", + "location": "Austin", + } + + _run_target_mode(ctx, ["Acme", "", ""]) + + args = ctx.hcatOllama.call_args[0] + assert args[0] == "1800" + assert args[1] == "/tmp/sha512.txt" + assert args[2] == "target" + assert set(args[3]) == {"company", "industry", "location"} + + +# --------------------------------------------------------------------------- +# main layer: hcatOllamaResearchTarget +# --------------------------------------------------------------------------- + + +class TestHcatOllamaResearchTarget: + def test_returns_researched_fields_and_shows_progress(self) -> None: + result = hc_main.llm.TargetResearchOutput(industry="mining", location="Perth") + with ( + mock.patch.object(hc_main, "ollamaAutoResearch", True), + mock.patch.object(hc_main, "ollamaModel", "qwen2.5:32b"), + mock.patch.object(hc_main, "ollamaTimeout", 30.0), + mock.patch.object(hc_main.llm, "research_target", return_value=result), + mock.patch.object(hc_main, "spinner") as spin, + ): + out = hc_main.hcatOllamaResearchTarget("Acme") + + assert out == {"industry": "mining", "location": "Perth"} + # The call must be wrapped in the progress spinner, not look like a hang. + assert spin.call_count == 1 + assert "Acme" in spin.call_args[0][0] + + def test_forwards_config_values_to_research_target(self) -> None: + result = hc_main.llm.TargetResearchOutput(industry="", location="") + with ( + mock.patch.object(hc_main, "ollamaAutoResearch", True), + mock.patch.object(hc_main, "ollamaUrl", "http://localhost:11434"), + mock.patch.object(hc_main, "ollamaModel", "m"), + mock.patch.object(hc_main, "ollamaNumCtx", 4096), + mock.patch.object(hc_main, "ollamaTimeout", 12.5), + mock.patch.object( + hc_main.llm, "research_target", return_value=result + ) as research, + ): + hc_main.hcatOllamaResearchTarget("Acme") + + assert research.call_args[0] == ("http://localhost:11434", "m", 4096, "Acme") + assert research.call_args.kwargs["timeout"] == 12.5 + + def test_timeout_degrades_to_blank_suggestions(self, capsys) -> None: + with ( + mock.patch.object(hc_main, "ollamaAutoResearch", True), + mock.patch.object(hc_main, "ollamaTimeout", 5.0), + mock.patch.object( + hc_main.llm, + "research_target", + side_effect=hc_main.llm.LLMTimeoutError("no response"), + ), + ): + out = hc_main.hcatOllamaResearchTarget("Acme") + + assert out == {"industry": "", "location": ""} + assert "timed out" in capsys.readouterr().out + + def test_connection_failure_degrades_to_blank_suggestions(self, capsys) -> None: + with ( + mock.patch.object(hc_main, "ollamaAutoResearch", True), + mock.patch.object( + hc_main.llm, + "research_target", + side_effect=ConnectionError("connection refused"), + ), + ): + out = hc_main.hcatOllamaResearchTarget("Acme") + + assert out == {"industry": "", "location": ""} + assert "unavailable" in capsys.readouterr().out + + def test_disabled_by_config_skips_the_model_call(self) -> None: + with ( + mock.patch.object(hc_main, "ollamaAutoResearch", False), + mock.patch.object(hc_main.llm, "research_target") as research, + ): + out = hc_main.hcatOllamaResearchTarget("Acme") + + research.assert_not_called() + assert out == {"industry": "", "location": ""} + + def test_blank_company_skips_the_model_call(self) -> None: + with ( + mock.patch.object(hc_main, "ollamaAutoResearch", True), + mock.patch.object(hc_main.llm, "research_target") as research, + ): + out = hc_main.hcatOllamaResearchTarget("") + + research.assert_not_called() + assert out == {"industry": "", "location": ""} + + def test_auto_research_default_is_enabled(self) -> None: + assert hc_main.ollamaAutoResearch is True From a75134274489bdebd2d926a688129a3e8605c833 Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Fri, 24 Jul 2026 18:51:48 -0400 Subject: [PATCH 10/11] style(ui): normalize interactive prompt decoration and default-value format MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove [*] prefix from all input() prompts (10 sites); [*] is a status/output marker and must not appear in user-facing input prompts. print() calls that use [*] are untouched. - Normalize default-value hints to the bare-parentheses form (7 instances already used it vs. 1 using "default N"); the lone outlier "\nEnter n-gram group size (default 3): " is rewritten to "(3)". - Add \n prefix to combipow "Add spaces between words?" prompt so it does not run flush against the last output line of the wordlist path loop. - All other prompts already satisfy: capital first letter, ends with ": ", and correct leading-whitespace context (tab-indented where the surrounding menu uses tabs, plain \n or none where it follows un-indented output). - No input() calls added or removed (verified: attacks.py 42→42, main.py 26→26). Co-Authored-By: Claude --- hate_crack/attacks.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/hate_crack/attacks.py b/hate_crack/attacks.py index c84ccf2..f1e704c 100644 --- a/hate_crack/attacks.py +++ b/hate_crack/attacks.py @@ -795,7 +795,7 @@ def combipow_crack(ctx: Any) -> None: _notify.prompt_notify_for_attack("Combipow") wordlist = None while wordlist is None: - path = input("\n[*] Enter path to wordlist (max 63 lines recommended): ").strip() + path = input("\nEnter path to wordlist (max 63 lines recommended): ").strip() if not path: continue if not os.path.isfile(path): @@ -813,7 +813,7 @@ def combipow_crack(ctx: Any) -> None: f"[*] Warning: {line_count} lines will generate a large number of combinations." ) wordlist = path - use_space_sep = input("[*] Add spaces between words? (Y/n): ").strip().lower() != "n" + use_space_sep = input("\nAdd spaces between words? (Y/n): ").strip().lower() != "n" ctx.hcatCombipow(ctx.hcatHashType, ctx.hcatHashFile, wordlist, use_space_sep) @@ -913,7 +913,7 @@ def ngram_attack(ctx: Any) -> None: print("No corpus selected. Aborting ngram attack.") return - group_size_raw = input("\nEnter n-gram group size (default 3): ").strip() + group_size_raw = input("\nEnter n-gram group size (3): ").strip() try: group_size = int(group_size_raw) if group_size_raw else 3 except ValueError: @@ -1118,8 +1118,8 @@ def wordlist_filter_length(ctx: Any) -> None: if not outfile: print("[!] Output path cannot be empty.") return - min_len = int(input("[*] Minimum length: ").strip() or "0") - max_len = int(input("[*] Maximum length: ").strip() or "0") + min_len = int(input("Minimum length: ").strip() or "0") + max_len = int(input("Maximum length: ").strip() or "0") if ctx.wordlist_filter_len(infile, outfile, min_len, max_len): print(f"\n[*] Filtered wordlist written to: {outfile}") else: @@ -1139,7 +1139,7 @@ def wordlist_filter_charclass_include(ctx: Any) -> None: print("[!] Output path cannot be empty.") return print("[*] Char class mask: 1=lowercase, 2=uppercase, 4=digit, 8=symbol (additive, e.g. 3=lower+upper)") - mask = int(input("[*] Enter mask value: ").strip() or "0") + mask = int(input("Mask value: ").strip() or "0") if ctx.wordlist_filter_req_include(infile, outfile, mask): print(f"\n[*] Filtered wordlist written to: {outfile}") else: @@ -1159,7 +1159,7 @@ def wordlist_filter_charclass_exclude(ctx: Any) -> None: print("[!] Output path cannot be empty.") return print("[*] Char class mask: 1=lowercase, 2=uppercase, 4=digit, 8=symbol (additive)") - mask = int(input("[*] Enter mask value: ").strip() or "0") + mask = int(input("Mask value: ").strip() or "0") if ctx.wordlist_filter_req_exclude(infile, outfile, mask): print(f"\n[*] Filtered wordlist written to: {outfile}") else: @@ -1178,8 +1178,8 @@ def wordlist_cut_substring(ctx: Any) -> None: if not outfile: print("[!] Output path cannot be empty.") return - offset = int(input("[*] Byte offset to start from: ").strip() or "0") - raw_length = input("[*] Length (leave blank for rest of line): ").strip() + offset = int(input("Byte offset to start from: ").strip() or "0") + raw_length = input("Length (leave blank for rest of line): ").strip() length = int(raw_length) if raw_length else None if ctx.wordlist_cutb(infile, outfile, offset, length): print(f"\n[*] Output written to: {outfile}") @@ -1211,7 +1211,7 @@ def wordlist_subtract_words(ctx: Any) -> None: print("\n[*] Subtract mode:") print(" 1. Single remove file (rli2 - faster for one file)") print(" 2. Multiple remove files (rli)") - mode = input("[*] Choose mode (1/2): ").strip() + mode = input("Choose mode (1/2): ").strip() if mode == "1": infile = ctx.select_file_with_autocomplete( @@ -1274,7 +1274,7 @@ def wordlist_shard(ctx: Any) -> None: if not outbase: print("[!] Output path cannot be empty.") return - mod = int(input("[*] Shard count (e.g. 4 to split into 4 parts): ").strip() or "0") + mod = int(input("Shard count (e.g. 4 to split into 4 parts): ").strip() or "0") if mod < 2: print("[!] Shard count must be at least 2.") return From 119dc29da462da563f92958e13f7113edb72e406 Mon Sep 17 00:00:00 2001 From: Justin Bollinger Date: Fri, 24 Jul 2026 18:53:39 -0400 Subject: [PATCH 11/11] docs: record UI polish and LLM mode additions under 2.12.0 2.12.0 is not yet tagged (latest tag is v2.11.4), so this work folds into that entry rather than cutting a new version. Co-Authored-By: Claude --- CHANGELOG.md | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 33c3e84..2c57ba1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,9 +22,37 @@ Dates are omitted for releases predating this file; see the git tags for exact t - **Wordlist (denylist) generation mode** for the LLM attack is now reachable from the menu: select the LLM attack (option 12), then choose "Wordlist" to derive basewords from a sample wordlist. +- **Cracked-password generation mode** for the LLM attack. Once a session has recovered + plaintexts, option 3 feeds them back to the model, which infers the organization's own + password conventions and generates new candidates in that style. Offered only when + `.out` has content, and it uses a dedicated prompt that tells the model not to + re-emit passwords already cracked. +- **Target research pre-fills the industry and location prompts.** In target mode, entering + the company name asks the local model to recall that organization's industry and location, + then offers them as editable defaults (Enter accepts, typing overrides). Values are + labelled as model guesses rather than verified OSINT, whitespace-collapsed, and capped at + 80 characters. Research runs entirely against the configured local Ollama server, so the + client name is never sent to a third party. Any failure or timeout falls back to blank + prompts and never blocks the attack. Disable with `ollamaAutoResearch: false`. +- **Live progress spinner** with an elapsed-seconds counter during Ollama generation, so a + model loading into VRAM is distinguishable from a hang. Automatically suppressed when + stdout is not a TTY. +- **`ollamaMaxSampleLines`** (default 500) caps how many sample passwords are sent to the + model, for both wordlist and cracked-password modes. ### Fixed +- **A large sample wordlist no longer stalls the LLM attack.** Wordlist mode read every line + into memory and pasted all of them into the prompt, so pointing it at `rockyou.txt` + materialized hundreds of megabytes and overran the model's context window — which looked + like a hang. The file is now streamed and evenly sampled across its whole length, and the + count actually used is reported (`Sampled 500 of 14,344,391 passwords from wordlist.`). +- **`HATE_CRACK_ARROW_MENU=1` now works in the LLM and OMEN submenus.** They hand-rolled + `print()` + `input()` instead of the shared menu helper, so arrow-key navigation silently + did nothing there. +- **A typo in a wordlist or generation-mode prompt no longer aborts the whole attack.** The + pickers and submenus re-prompt instead of dropping back to the main menu, and offer an + explicit cancel. - **The LLM attack no longer hangs forever waiting on Ollama.** Generation requests are now bounded by a configurable timeout (`ollamaTimeout` in `config.json`, default 300 seconds). Previously, if Ollama accepted the connection but never replied — most commonly a large