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 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 diff --git a/README.md b/README.md index 9c24748..32e2738 100644 --- a/README.md +++ b/README.md @@ -472,9 +472,36 @@ 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 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 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. + + 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. ### Notifications (menu option 82) @@ -824,6 +851,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/config.json.example b/config.json.example index 9efabe3..ec89e9a 100644 --- a/config.json.example +++ b/config.json.example @@ -28,6 +28,8 @@ "ollamaModel": "qwen2.5:32b", "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 b63b17c..f1e704c 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,35 +511,99 @@ 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") - print("\n\tLLM Attack") - print("\t1. Target info (company / industry / location)") - print("\t2. Wordlist (generate basewords from a sample wordlist)") - choice = input("\n\tSelect generation mode: ").strip() + # 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 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 - ctx.hcatOllama(ctx.hcatHashType, ctx.hcatHashFile, "wordlist", path) - else: - print("\t[!] Invalid selection.") + if choice == "1": + company = input("Company name: ").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, + "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 + 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.""" + """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) @@ -549,18 +614,21 @@ def _omen_pick_training_wordlist(ctx: Any, title: str = "Training Wordlists"): 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 + print("\tq. Cancel") + while True: + 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: @@ -582,16 +650,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.") @@ -618,11 +690,12 @@ 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) + # Print the grid once, outside the retry loop — see _omen_pick_training_wordlist. entries = [] if has_cracked: entries.append("0) Cracked passwords (current session)") @@ -636,20 +709,23 @@ def _markov_pick_training_source(ctx: Any): 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 + print("\tq. Cancel") + while True: + 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: @@ -719,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): @@ -737,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) @@ -837,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: @@ -1042,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: @@ -1063,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: @@ -1083,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: @@ -1102,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}") @@ -1135,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( @@ -1198,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 diff --git a/hate_crack/config.json.example b/hate_crack/config.json.example index c20f6f8..fe8ffb8 100644 --- a/hate_crack/config.json.example +++ b/hate_crack/config.json.example @@ -22,6 +22,11 @@ "hashview_url": "http://localhost:8443", "hashview_api_key": "", "hashmob_api_key": "", + "ollamaModel": "qwen2.5:32b", + "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 184d4b0..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 " @@ -74,6 +132,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,9 +182,81 @@ 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}") +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, @@ -117,11 +278,9 @@ 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, - ) - prompt_generator = _TARGET_PROMPT if mode == "target" else _WORDLIST_PROMPT + client = _build_client(url, timeout) + # _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 b801441..f6dafd0 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,8 @@ 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)) +ollamaAutoResearch = bool(config_parser.get("ollamaAutoResearch", True)) omenTrainingList = config_parser.get("omenTrainingList", "rockyou.txt") omenMaxCandidates = int(config_parser.get("omenMaxCandidates", 1000000)) @@ -895,6 +898,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. @@ -2036,6 +2056,111 @@ 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 + + +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" @@ -2046,23 +2171,31 @@ def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): if not os.path.isfile(wordlist_path): print(f"Error: Wordlist not found: {wordlist_path}") return - lines = [] - try: - 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) - except Exception as e: - print(f"Error reading wordlist: {e}") + + sampled = _sample_plaintext_file(wordlist_path, ollamaMaxSampleLines) + if sampled is None: return - print(f"Loaded {len(lines)} passwords from wordlist.") - gen_context = {"sample": "\n".join(lines)} + 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 + + 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 else: @@ -2070,16 +2203,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..cc63d54 --- /dev/null +++ b/hate_crack/progress.py @@ -0,0 +1,71 @@ +"""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() + # 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() diff --git a/tests/test_attacks_behavior.py b/tests/test_attacks_behavior.py index 2ce1bab..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,24 +385,234 @@ 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 + ) -> 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)) + + 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) + + # 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 + ) -> 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)) + + 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) + + 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 + ) -> 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)) + + 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) + + 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( + 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("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_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..1e8ef94 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( @@ -145,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 diff --git a/tests/test_ollama_wordlist_sampling.py b/tests/test_ollama_wordlist_sampling.py new file mode 100644 index 0000000..d4e193a --- /dev/null +++ b/tests/test_ollama_wordlist_sampling.py @@ -0,0 +1,461 @@ +"""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(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 + + +# --------------------------------------------------------------------------- +# 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" + + +# --------------------------------------------------------------------------- +# 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 + + +# --------------------------------------------------------------------------- +# Invalid-cap guard — zero/negative ollamaMaxSampleLines 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 +# --------------------------------------------------------------------------- + + +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 + + +# --------------------------------------------------------------------------- +# _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:") == "" + + +# --------------------------------------------------------------------------- +# 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 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 diff --git a/tests/test_progress_spinner.py b/tests/test_progress_spinner.py new file mode 100644 index 0000000..c54c68b --- /dev/null +++ b/tests/test_progress_spinner.py @@ -0,0 +1,151 @@ +"""Tests for hate_crack/progress.py — the generic spinner context manager.""" + +from __future__ import annotations + +import io +import re +import sys +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( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """In a non-TTY environment the message is printed once and no thread runs.""" + 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 must never construct a Thread.""" + buf = io.StringIO() + buf.isatty = lambda: False # type: ignore[attr-defined] + + with mock.patch("sys.stdout", buf), mock.patch( + "hate_crack.progress.threading.Thread" + ) as mock_thread: + with spinner("No threads please"): + pass + mock_thread.assert_not_called() + + +# --------------------------------------------------------------------------- +# 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..."): + 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. + assert "\033[2K\r" in output or ("\033[2K" in output) + + +def test_tty_shows_elapsed_seconds() -> None: + """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] + + # 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..."): + pass # body exits immediately; the thread gets one tick then stops + + output = buf.getvalue() + 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: + """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") + + +# --------------------------------------------------------------------------- +# 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" + )