diff --git a/.git-blame-ignore-revs b/.git-blame-ignore-revs new file mode 100644 index 0000000..8299e92 --- /dev/null +++ b/.git-blame-ignore-revs @@ -0,0 +1,9 @@ +# Revisions listed here are skipped by `git blame --ignore-revs-file`. +# Enable locally with: +# git config blame.ignoreRevsFile .git-blame-ignore-revs +# +# Only add commits whose changes are purely mechanical (e.g., formatter +# passes, mass-rename refactors). Do not add feature commits. + +# style: ruff format pass for Notifications submenu +9b684bb44c3cd04b00a5421691179ea56162780c diff --git a/README.md b/README.md index 6369899..19c75e6 100644 --- a/README.md +++ b/README.md @@ -472,6 +472,23 @@ The LLM Attack (option 15) uses Ollama to generate password candidates. Configur - **`ollamaNumCtx`** — Context window size for the model (default: `2048`). - The Ollama URL defaults to `http://localhost:11434`. Ensure Ollama is running before using the LLM Attack. +### Notifications (menu option 82) + +hate_crack can send Pushover push notifications when attacks complete and, +optionally, when individual hashes are cracked. All controls live under +main-menu option `82 — Notifications`: + +1. **Toggle Pushover Notifications [ON/OFF]** — master switch. Persists to `config.json` as `notify_enabled`. +2. **Toggle Per-Crack Notifications [ON/OFF]** — when ON, a background tailer watches the `.out` file and pushes a notification per crack (with per-tick burst aggregation). Persists to `config.json` as `notify_per_crack_enabled`. Cannot be enabled while the master switch is OFF — enable option 1 first. +3. **Send Test Pushover Notification** — fires a canned push so you can confirm your Pushover token/user pair works. Works even when the master switch is OFF. + +Credentials and tuning knobs remain config-file-only in `config.json`: + +- `notify_pushover_token`, `notify_pushover_user` — required for any push to fire. +- `notify_attack_allowlist` — attack names that auto-consent without the `[y/N/always]` prompt. Populated automatically when you answer `always`. +- `notify_suppress_in_orchestrators` (default `true`) — silences nested attacks launched by Quick/Extensive/Brute-Force wrappers; the wrapper fires a single summary instead. +- `notify_max_cracks_per_burst` (default `5`), `notify_poll_interval_seconds` (default `5.0`) — per-crack tailer tuning. See `hate_crack/notify/tailer.py` for the burst aggregation logic. + ### Wordlist Tools (menu option 80) The Wordlist Tools submenu provides 7 wordlist preprocessing utilities backed by hashcat-utils binaries. Access via option **80** in the main menu. diff --git a/config.json.example b/config.json.example index 78d3b36..27bf100 100644 --- a/config.json.example +++ b/config.json.example @@ -37,5 +37,13 @@ "hcatCombinator3", "hcatCombinatorX", "hcatHybrid", "hcatYoloCombination", "hcatMiddleCombinator", "hcatThoroughCombinator", "hcatCombipow", "hcatPrince", "hcatPermute" - ] + ], + "notify_enabled": false, + "notify_pushover_token": "", + "notify_pushover_user": "", + "notify_per_crack_enabled": false, + "notify_attack_allowlist": [], + "notify_suppress_in_orchestrators": true, + "notify_max_cracks_per_burst": 5, + "notify_poll_interval_seconds": 5.0 } diff --git a/hate_crack.py b/hate_crack.py index 3895f4b..178655d 100755 --- a/hate_crack.py +++ b/hate_crack.py @@ -94,6 +94,7 @@ def get_main_menu_options(): "22": _attacks.combipow_crack, "80": _attacks.wordlist_tools_submenu, "81": _attacks.rule_tools_submenu, + "82": notifications_submenu, "90": download_hashmob_rules, "91": weakpass_wordlist_menu, "92": download_hashmob_wordlists, diff --git a/hate_crack/attacks.py b/hate_crack/attacks.py index 920eca1..9a26d32 100644 --- a/hate_crack/attacks.py +++ b/hate_crack/attacks.py @@ -4,6 +4,7 @@ import os import readline 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.menu import interactive_menu @@ -107,6 +108,7 @@ def _select_rules(ctx) -> list[str] | None: def quick_crack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Quick Crack") wordlist_choice = None default_dir = ctx.hcatOptimizedWordlists @@ -177,6 +179,7 @@ def quick_crack(ctx: Any) -> None: def loopback_attack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Loopback") empty_wordlist = os.path.join(ctx.hcatWordlists, "empty.txt") os.makedirs(ctx.hcatWordlists, exist_ok=True) if not os.path.exists(empty_wordlist): @@ -200,26 +203,43 @@ def loopback_attack(ctx: Any) -> None: def extensive_crack(ctx: Any) -> None: - ctx.hcatBruteForce(ctx.hcatHashType, ctx.hcatHashFile, "1", "7") - ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatBruteCount) - ctx.hcatDictionary(ctx.hcatHashType, ctx.hcatHashFile) - ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatDictionaryCount) - hcatTargetTime = 4 * 60 * 60 - ctx.hcatTopMask(ctx.hcatHashType, ctx.hcatHashFile, hcatTargetTime) - ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatMaskCount) - ctx.hcatFingerprint( - ctx.hcatHashType, ctx.hcatHashFile, 7, run_hybrid_on_expanded=False + # Orchestrator attack: chains ~14 primitives. We suppress each primitive's + # own notifications and fire exactly one "Extensive Crack complete" at the + # end with the aggregate delta. This both prevents notification spam and + # gives the user an actually-useful summary. + _notify.prompt_notify_for_attack("Extensive Crack") + out_path = ctx.hcatHashFile + ".out" + cracked_before = ctx.lineCount(out_path) if os.path.exists(out_path) else 0 + with _notify.suppressed_notifications(): + ctx.hcatBruteForce(ctx.hcatHashType, ctx.hcatHashFile, "1", "7") + ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatBruteCount) + ctx.hcatDictionary(ctx.hcatHashType, ctx.hcatHashFile) + ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatDictionaryCount) + hcatTargetTime = 4 * 60 * 60 + ctx.hcatTopMask(ctx.hcatHashType, ctx.hcatHashFile, hcatTargetTime) + ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatMaskCount) + ctx.hcatFingerprint( + ctx.hcatHashType, ctx.hcatHashFile, 7, run_hybrid_on_expanded=False + ) + ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatFingerprintCount) + ctx.hcatCombination(ctx.hcatHashType, ctx.hcatHashFile) + ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatCombinationCount) + ctx.hcatHybrid(ctx.hcatHashType, ctx.hcatHashFile) + ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatHybridCount) + ctx.hcatGoodMeasure(ctx.hcatHashType, ctx.hcatHashFile) + ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatExtraCount) + cracked_after = ctx.lineCount(out_path) if os.path.exists(out_path) else 0 + _notify.notify_job_done( + "Extensive Crack", cracked_after, ctx.hcatHashFile ) - ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatFingerprintCount) - ctx.hcatCombination(ctx.hcatHashType, ctx.hcatHashFile) - ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatCombinationCount) - ctx.hcatHybrid(ctx.hcatHashType, ctx.hcatHashFile) - ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatHybridCount) - ctx.hcatGoodMeasure(ctx.hcatHashType, ctx.hcatHashFile) - ctx.hcatRecycle(ctx.hcatHashType, ctx.hcatHashFile, ctx.hcatExtraCount) + # Note: ``cracked_before`` is tracked for potential future per-orchestrator + # delta reporting, but today the notify message uses the absolute count + # because that matches what single-attack notifications already report. + _ = cracked_before def brute_force_crack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Brute Force") hcatMinLen = int( input("\nEnter the minimum password length to brute force (1): ") or 1 ) @@ -230,6 +250,7 @@ def brute_force_crack(ctx: Any) -> None: def top_mask_crack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Top Mask") hcatTargetTime = int( input("\nEnter a target time for completion in hours (4): ") or 4 ) @@ -238,6 +259,7 @@ def top_mask_crack(ctx: Any) -> None: def fingerprint_crack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Fingerprint") while True: raw = input("\nEnter expander max length (7-36) (7): ").strip() if raw == "": @@ -261,6 +283,7 @@ def fingerprint_crack(ctx: Any) -> None: def combinator_crack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Combinator") print("\n" + "=" * 60) print("COMBINATOR ATTACK") print("=" * 60) @@ -299,6 +322,7 @@ def combinator_crack(ctx: Any) -> None: def hybrid_crack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Hybrid") print("\n" + "=" * 60) print("HYBRID ATTACK") print("=" * 60) @@ -367,22 +391,27 @@ def hybrid_crack(ctx: Any) -> None: def pathwell_crack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Pathwell Brute Force") ctx.hcatPathwellBruteForce(ctx.hcatHashType, ctx.hcatHashFile) def prince_attack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("PRINCE") ctx.hcatPrince(ctx.hcatHashType, ctx.hcatHashFile) def yolo_combination(ctx: Any) -> None: + _notify.prompt_notify_for_attack("YOLO Combination") ctx.hcatYoloCombination(ctx.hcatHashType, ctx.hcatHashFile) def thorough_combinator(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Thorough Combinator") ctx.hcatThoroughCombinator(ctx.hcatHashType, ctx.hcatHashFile) def middle_combinator(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Middle Combinator") ctx.hcatMiddleCombinator(ctx.hcatHashType, ctx.hcatHashFile) @@ -447,10 +476,12 @@ def combinator_3plus_crack(ctx: Any) -> None: def bandrel_method(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Bandrel") ctx.hcatBandrel(ctx.hcatHashType, ctx.hcatHashFile) def ollama_attack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("LLM") print("\n\tLLM Attack") company = input("Company name: ").strip() industry = input("Industry: ").strip() @@ -491,6 +522,7 @@ def _omen_pick_training_wordlist(ctx: Any): def omen_attack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("OMEN") print("\n\tOMEN Attack (Ordered Markov ENumerator)") omen_dir = os.path.join(ctx.hate_path, "omen") create_bin = os.path.join(omen_dir, ctx.hcatOmenCreateBin) @@ -579,6 +611,7 @@ def _markov_pick_training_source(ctx: Any): def adhoc_mask_crack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Ad-hoc Mask") print( "\nEnter a hashcat mask. Tokens: ?l=lower ?u=upper ?d=digit ?s=special ?a=all ?b=binary ?1-?4=custom" ) @@ -603,6 +636,7 @@ def adhoc_mask_crack(ctx: Any) -> None: def markov_brute_force(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Markov Brute Force") print("\n\tMarkov Brute Force Attack") hcstat2_path = f"{ctx.hcatHashFile}.hcstat2" need_training = True @@ -640,6 +674,7 @@ def markov_brute_force(ctx: Any) -> None: 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() @@ -665,6 +700,7 @@ def combipow_crack(ctx: Any) -> None: def generate_rules_crack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Random Rules") print("\n" + "=" * 60) print("RANDOM RULES ATTACK") print("=" * 60) @@ -743,6 +779,7 @@ def generate_rules_crack(ctx: Any) -> None: def ngram_attack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("N-gram") print("\n" + "=" * 60) print("NGRAM ATTACK") print("=" * 60) @@ -769,6 +806,7 @@ def ngram_attack(ctx: Any) -> None: def permute_crack(ctx: Any) -> None: + _notify.prompt_notify_for_attack("Permute") print("\n" + "=" * 60) print("PERMUTATION ATTACK") print("=" * 60) diff --git a/hate_crack/config.json.example b/hate_crack/config.json.example index 013440b..c20f6f8 100644 --- a/hate_crack/config.json.example +++ b/hate_crack/config.json.example @@ -25,5 +25,13 @@ "passgptModel": "javirandor/passgpt-10characters", "passgptMaxCandidates": 1000000, "passgptBatchSize": 1024, - "passgptTrainingList": "" + "passgptTrainingList": "", + "notify_enabled": false, + "notify_pushover_token": "", + "notify_pushover_user": "", + "notify_per_crack_enabled": false, + "notify_attack_allowlist": [], + "notify_suppress_in_orchestrators": true, + "notify_max_cracks_per_burst": 5, + "notify_poll_interval_seconds": 5.0 } diff --git a/hate_crack/main.py b/hate_crack/main.py index 6242794..08800d5 100755 --- a/hate_crack/main.py +++ b/hate_crack/main.py @@ -401,7 +401,9 @@ if hcatOptimizedWordlists: if not os.path.isdir(hcatOptimizedWordlists): fallback_optimized = os.path.join(hate_path, "optimized_wordlists") if os.path.isdir(fallback_optimized): - print(f"[!] hcatOptimizedWordlists directory not found: {hcatOptimizedWordlists}") + print( + f"[!] hcatOptimizedWordlists directory not found: {hcatOptimizedWordlists}" + ) print(f"[!] Falling back to {fallback_optimized}") hcatOptimizedWordlists = fallback_optimized else: @@ -417,8 +419,12 @@ pipalPath = config_parser["pipalPath"] hcatDictionaryWordlist = config_parser["hcatDictionaryWordlist"] hcatHybridlist = config_parser["hcatHybridlist"] hcatCombinationWordlist = config_parser["hcatCombinationWordlist"] -hcatCombinator3Wordlist = config_parser.get("hcatCombinator3Wordlist", ["rockyou.txt", "rockyou.txt", "rockyou.txt"]) -hcatCombinatorXWordlist = config_parser.get("hcatCombinatorXWordlist", ["rockyou.txt", "rockyou.txt"]) +hcatCombinator3Wordlist = config_parser.get( + "hcatCombinator3Wordlist", ["rockyou.txt", "rockyou.txt", "rockyou.txt"] +) +hcatCombinatorXWordlist = config_parser.get( + "hcatCombinatorXWordlist", ["rockyou.txt", "rockyou.txt"] +) hcatMiddleCombinatorMasks = config_parser["hcatMiddleCombinatorMasks"] hcatMiddleBaseList = config_parser["hcatMiddleBaseList"] hcatThoroughCombinatorMasks = config_parser["hcatThoroughCombinatorMasks"] @@ -443,6 +449,14 @@ except KeyError: pass check_for_updates_enabled = config_parser.get("check_for_updates", True) +# Notification subsystem bootstrap. The notify module stores its own +# settings snapshot; we just hand it the resolved config path so it can +# atomically rewrite config.json when the user toggles enabled / answers +# "always" at a prompt. +from hate_crack import notify as _notify # noqa: E402 (kept close to config load) + +_notify.init(_config_path, config_parser) + hcatExpanderBin = "expander.bin" hcatCombinatorBin = "combinator.bin" hcatPrinceBin = "pp64.bin" @@ -576,8 +590,12 @@ hcatDictionaryWordlist = _normalize_wordlist_setting( hcatCombinationWordlist = _normalize_wordlist_setting( hcatCombinationWordlist, wordlists_dir ) -hcatCombinator3Wordlist = _normalize_wordlist_setting(hcatCombinator3Wordlist, wordlists_dir) -hcatCombinatorXWordlist = _normalize_wordlist_setting(hcatCombinatorXWordlist, wordlists_dir) +hcatCombinator3Wordlist = _normalize_wordlist_setting( + hcatCombinator3Wordlist, wordlists_dir +) +hcatCombinatorXWordlist = _normalize_wordlist_setting( + hcatCombinatorXWordlist, wordlists_dir +) hcatHybridlist = _normalize_wordlist_setting(hcatHybridlist, wordlists_dir) hcatMiddleBaseList = _normalize_wordlist_setting(hcatMiddleBaseList, wordlists_dir) hcatThoroughBaseList = _normalize_wordlist_setting(hcatThoroughBaseList, wordlists_dir) @@ -729,6 +747,93 @@ def _debug_cmd(cmd): print(f"[DEBUG] hashcat cmd: {_format_cmd(cmd)}") +def _run_hcat_cmd( + cmd, + attack_name: str = "", + hash_file: str | None = None, + *, + stdin=None, + companion_procs=None, + reraise_interrupt: bool = False, + out_path: str | None = None, +): + """Execute a hashcat subprocess and bracket it with notify hooks. + + This consolidates the ``hcatProcess = subprocess.Popen(cmd); try: + wait() except KeyboardInterrupt: kill()`` dance that was duplicated + at ~31 sites in this module. The payoff: every hashcat invocation + now fires job-done notifications consistently, and the per-crack + tailer lifecycle is handled in exactly one place. + + - ``attack_name`` is the label that appears in notifications. Pass + an empty string for no-notify invocations. + - ``hash_file`` is required to locate ``{hash_file}.out`` for the + tailer. When omitted, we skip the tailer and the job-done count. + - ``stdin`` mirrors the ``subprocess.Popen(..., stdin=...)`` kwarg + for generator-pipe callers. + - ``companion_procs`` is a list of generator ``Popen`` handles that + feed into this hashcat instance. On normal completion we + ``wait()`` them; on ``KeyboardInterrupt`` we ``kill()`` them + alongside the hashcat process. This preserves the prior behavior + where a ctrl-C must tear down both sides of a pipe. + + Notifications are fire-and-forget: suppression (see + ``notify.suppressed_notifications``) and disabled-globally state are + both handled inside the notify module, so callers need not branch. + """ + global hcatProcess + + companions = list(companion_procs) if companion_procs else [] + + # Resolve the output file path used for the tailer and cracked-count + # readback. Most hashcat calls write to ``{hash_file}.out``; a few + # multi-phase flows (LM-to-NT) write to a different file, in which + # case the caller passes ``out_path`` explicitly. + resolved_out = out_path if out_path else (hash_file + ".out" if hash_file else None) + + tailer = None + if attack_name and resolved_out and not _notify.is_suppressed(): + tailer = _notify.start_tailer(resolved_out, attack_name) + + popen_kwargs = {"stdin": stdin} if stdin is not None else {} + hcatProcess = subprocess.Popen(cmd, **popen_kwargs) + interrupted = False + try: + hcatProcess.wait() + for gen in companions: + try: + gen.wait() + except Exception: + pass + except KeyboardInterrupt: + interrupted = True + print("Killing PID {0}...".format(str(hcatProcess.pid))) + hcatProcess.kill() + for gen in companions: + try: + gen.kill() + except Exception: + pass + finally: + _notify.stop_tailer(tailer) + + # Only incur a lineCount read when notifications will actually fire. + # This avoids disturbing existing tests that assert a specific number + # of file reads during an attack; ``_should_fire`` mirrors the check + # inside ``notify_job_done`` itself. + if ( + attack_name + and resolved_out + and not _notify.is_suppressed() + and _notify.get_settings().enabled + ): + cracked = lineCount(resolved_out) + _notify.notify_job_done(attack_name, cracked, hash_file or resolved_out) + + if interrupted and reraise_interrupt: + raise KeyboardInterrupt + + def _is_gzipped(path: str) -> bool: try: with open(path, "rb") as f: @@ -1189,12 +1294,7 @@ def hcatBruteForce(hcatHashType, hcatHashFile, hcatMinLen, hcatMaxLen): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Brute Force", hash_file=hcatHashFile) hcatBruteCount = lineCount(hcatHashFile + ".out") @@ -1226,12 +1326,7 @@ def hcatDictionary(hcatHashType, hcatHashFile): cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) cmd = _add_debug_mode_for_rules(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Dictionary", hash_file=hcatHashFile) rule_d3ad0ne = get_rule_path("d3ad0ne.rule") rule_toxic = get_rule_path("T0XlC.rule") @@ -1267,12 +1362,7 @@ def hcatDictionary(hcatHashType, hcatHashFile): cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) cmd = _add_debug_mode_for_rules(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Dictionary", hash_file=hcatHashFile) finally: os.unlink(combined_path) @@ -1316,12 +1406,7 @@ def hcatQuickDictionary( ) cmd = _add_debug_mode_for_rules(cmd) _debug_cmd(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Quick Dictionary", hash_file=hcatHashFile) # Top Mask Attack @@ -1383,12 +1468,7 @@ def hcatTopMask(hcatHashType, hcatHashFile, hcatTargetTime): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Top Mask", hash_file=hcatHashFile) hcatMaskCount = lineCount(hcatHashFile + ".out") - hcatHashCracked @@ -1462,12 +1542,9 @@ def hcatFingerprint( _insert_optimized_flag(fingerprint_cmd) fingerprint_cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(fingerprint_cmd) - hcatProcess = subprocess.Popen(fingerprint_cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd( + fingerprint_cmd, attack_name="Fingerprint", hash_file=hcatHashFile + ) # Secondary attack: run hybrid on the expanded candidates (mode 6/7 variants). # This is intentionally optional to avoid changing the "extensive" pipeline ordering. @@ -1529,12 +1606,7 @@ def hcatCombination(hcatHashType, hcatHashFile, wordlists=None): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Combination", hash_file=hcatHashFile) hcatCombinationCount = lineCount(hcatHashFile + ".out") - hcatHashCracked @@ -1568,15 +1640,15 @@ def hcatCombinator3(hcatHashType, hcatHashFile, wordlists): _append_potfile_arg(hashcat_cmd) generator_proc = subprocess.Popen(generator_cmd, stdout=subprocess.PIPE) assert generator_proc.stdout is not None - hcatProcess = subprocess.Popen(hashcat_cmd, stdin=generator_proc.stdout) - generator_proc.stdout.close() - try: - hcatProcess.wait() - generator_proc.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() - generator_proc.kill() + _run_hcat_cmd( + hashcat_cmd, + attack_name="Combinator3", + hash_file=hcatHashFile, + stdin=generator_proc.stdout, + companion_procs=[generator_proc], + ) + if generator_proc.stdout: + generator_proc.stdout.close() hcatCombinator3Count = lineCount(hcatHashFile + ".out") - hcatHashCracked @@ -1614,15 +1686,15 @@ def hcatCombinatorX(hcatHashType, hcatHashFile, wordlists, separator=None): _append_potfile_arg(hashcat_cmd) generator_proc = subprocess.Popen(generator_cmd, stdout=subprocess.PIPE) assert generator_proc.stdout is not None - hcatProcess = subprocess.Popen(hashcat_cmd, stdin=generator_proc.stdout) - generator_proc.stdout.close() - try: - hcatProcess.wait() - generator_proc.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() - generator_proc.kill() + _run_hcat_cmd( + hashcat_cmd, + attack_name="CombinatorX", + hash_file=hcatHashFile, + stdin=generator_proc.stdout, + companion_procs=[generator_proc], + ) + if generator_proc.stdout: + generator_proc.stdout.close() hcatCombinatorXCount = lineCount(hcatHashFile + ".out") - hcatHashCracked @@ -1649,15 +1721,15 @@ def hcatNgramX(hcatHashType, hcatHashFile, corpus, group_size=3): _append_potfile_arg(hashcat_cmd) generator_proc = subprocess.Popen(generator_cmd, stdout=subprocess.PIPE) assert generator_proc.stdout is not None - hcatProcess = subprocess.Popen(hashcat_cmd, stdin=generator_proc.stdout) - generator_proc.stdout.close() - try: - hcatProcess.wait() - generator_proc.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() - generator_proc.kill() + _run_hcat_cmd( + hashcat_cmd, + attack_name="NgramX", + hash_file=hcatHashFile, + stdin=generator_proc.stdout, + companion_procs=[generator_proc], + ) + if generator_proc.stdout: + generator_proc.stdout.close() hcatNgramXCount = lineCount(hcatHashFile + ".out") - hcatHashCracked @@ -1711,12 +1783,7 @@ def hcatHybrid(hcatHashType, hcatHashFile, wordlists=None): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Hybrid", hash_file=hcatHashFile) hcatHybridCount = lineCount(hcatHashFile + ".out") - hcatHashCracked @@ -1749,13 +1816,12 @@ def hcatYoloCombination(hcatHashType, hcatHashFile): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() - raise + _run_hcat_cmd( + cmd, + attack_name="YOLO Combination", + hash_file=hcatHashFile, + reraise_interrupt=True, + ) except KeyboardInterrupt: pass @@ -1800,12 +1866,7 @@ def hcatBandrel(hcatHashType, hcatHashFile): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Bandrel", hash_file=hcatHashFile) print( "Checking passwords against pipal for top {0} passwords and basewords".format( pipal_count @@ -1845,12 +1906,7 @@ def hcatBandrel(hcatHashType, hcatHashFile): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Bandrel", hash_file=hcatHashFile) # Pull an Ollama model via the /api/pull streaming endpoint @@ -2053,12 +2109,14 @@ def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): ] cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) try: - hcatProcess.wait() + _run_hcat_cmd( + cmd, + attack_name="LLM", + hash_file=hcatHashFile, + reraise_interrupt=True, + ) except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() return # Step D: Run hashcat with LLM candidates against every rule in the rules directory @@ -2088,12 +2146,14 @@ def hcatOllama(hcatHashType, hcatHashFile, mode, context_data): ] cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) try: - hcatProcess.wait() + _run_hcat_cmd( + cmd, + attack_name="LLM", + hash_file=hcatHashFile, + reraise_interrupt=True, + ) except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() return @@ -2135,13 +2195,12 @@ def hcatMiddleCombinator(hcatHashType, hcatHashFile): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() - raise + _run_hcat_cmd( + cmd, + attack_name="Middle Combinator", + hash_file=hcatHashFile, + reraise_interrupt=True, + ) except KeyboardInterrupt: pass @@ -2180,12 +2239,7 @@ def hcatThoroughCombinator(hcatHashType, hcatHashFile): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Thorough Combinator", hash_file=hcatHashFile) try: for x in range(len(masks)): @@ -2209,13 +2263,12 @@ def hcatThoroughCombinator(hcatHashType, hcatHashFile): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() - raise + _run_hcat_cmd( + cmd, + attack_name="Thorough Combinator", + hash_file=hcatHashFile, + reraise_interrupt=True, + ) except KeyboardInterrupt: pass try: @@ -2240,13 +2293,12 @@ def hcatThoroughCombinator(hcatHashType, hcatHashFile): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() - raise + _run_hcat_cmd( + cmd, + attack_name="Thorough Combinator", + hash_file=hcatHashFile, + reraise_interrupt=True, + ) except KeyboardInterrupt: pass try: @@ -2273,11 +2325,14 @@ def hcatThoroughCombinator(hcatHashType, hcatHashFile): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - hcatProcess.wait() + _run_hcat_cmd( + cmd, + attack_name="Thorough Combinator", + hash_file=hcatHashFile, + reraise_interrupt=True, + ) except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + pass # Pathwell Mask Brute Force Attack @@ -2300,12 +2355,7 @@ def hcatPathwellBruteForce(hcatHashType, hcatHashFile): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Pathwell Brute Force", hash_file=hcatHashFile) def hcatAdHocMask(hcatHashType, hcatHashFile, mask, custom_charsets=""): @@ -2329,12 +2379,7 @@ def hcatAdHocMask(hcatHashType, hcatHashFile, mask, custom_charsets=""): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Ad-hoc Mask", hash_file=hcatHashFile) def hcatMarkovTrain(source_file, hcatHashFile): @@ -2362,8 +2407,14 @@ def hcatMarkovTrain(source_file, hcatHashFile): hcatProcess.wait(timeout=300) if hcatProcess.returncode != 0: _, stderr_data = hcatProcess.communicate() - err_msg = stderr_data.decode("utf-8", errors="replace") if stderr_data else "Unknown error" - print(f"[!] hcstat2gen.bin failed with code {hcatProcess.returncode}: {err_msg}") + err_msg = ( + stderr_data.decode("utf-8", errors="replace") + if stderr_data + else "Unknown error" + ) + print( + f"[!] hcstat2gen.bin failed with code {hcatProcess.returncode}: {err_msg}" + ) return False except subprocess.TimeoutExpired: print("[!] hcstat2gen.bin timed out after 300 seconds") @@ -2429,12 +2480,7 @@ def hcatMarkovBruteForce(hcatHashType, hcatHashFile, hcatMinLen, hcatMaxLen): _insert_optimized_flag(cmd) cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Markov Brute Force", hash_file=hcatHashFile) # Combipow Passphrase Attack @@ -2478,15 +2524,16 @@ def hcatCombipow(hcatHashType, hcatHashFile, wordlist, use_space_sep=True): hashcat_cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(hashcat_cmd) generator_proc = subprocess.Popen(generator_cmd, stdout=subprocess.PIPE) - hcatProcess = subprocess.Popen(hashcat_cmd, stdin=generator_proc.stdout) - generator_proc.stdout.close() try: - hcatProcess.wait() - generator_proc.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() - generator_proc.kill() + _run_hcat_cmd( + hashcat_cmd, + attack_name="Combipow", + hash_file=hcatHashFile, + stdin=generator_proc.stdout, + companion_procs=[generator_proc], + ) + if generator_proc.stdout: + generator_proc.stdout.close() finally: if tmp_file is not None: with contextlib.suppress(OSError): @@ -2532,15 +2579,15 @@ def hcatPrince(hcatHashType, hcatHashFile): hashcat_cmd = _add_debug_mode_for_rules(hashcat_cmd) with _open_wordlist(prince_base) as base: prince_proc = subprocess.Popen(prince_cmd, stdin=base, stdout=subprocess.PIPE) - hcatProcess = subprocess.Popen(hashcat_cmd, stdin=prince_proc.stdout) - prince_proc.stdout.close() - try: - hcatProcess.wait() - prince_proc.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() - prince_proc.kill() + _run_hcat_cmd( + hashcat_cmd, + attack_name="PRINCE", + hash_file=hcatHashFile, + stdin=prince_proc.stdout, + companion_procs=[prince_proc], + ) + if prince_proc.stdout: + prince_proc.stdout.close() def hcatPermute(hcatHashType, hcatHashFile, wordlist): @@ -2570,17 +2617,15 @@ def hcatPermute(hcatHashType, hcatHashFile, wordlist): permute_proc = subprocess.Popen( [permute_path], stdin=wl_file, stdout=subprocess.PIPE ) - hcatProcess = subprocess.Popen( - hashcat_cmd, stdin=permute_proc.stdout + _run_hcat_cmd( + hashcat_cmd, + attack_name="Permute", + hash_file=hcatHashFile, + stdin=permute_proc.stdout, + companion_procs=[permute_proc], ) - permute_proc.stdout.close() - try: - hcatProcess.wait() - permute_proc.wait() - except KeyboardInterrupt: - print(f"Killing PID {hcatProcess.pid}...") - hcatProcess.kill() - permute_proc.kill() + if permute_proc.stdout: + permute_proc.stdout.close() hcatPermuteCount = lineCount(f"{hcatHashFile}.out") - hcatHashCracked @@ -2709,16 +2754,21 @@ def hcatOmen(hcatHashType, hcatHashFile, max_candidates, hcatChains=""): enum_proc = subprocess.Popen( enum_cmd, cwd=model_dir, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) - hcatProcess = subprocess.Popen(hashcat_cmd, stdin=enum_proc.stdout) - enum_proc.stdout.close() try: - hcatProcess.wait() - enum_proc.wait() + _run_hcat_cmd( + hashcat_cmd, + attack_name="OMEN", + hash_file=hcatHashFile, + stdin=enum_proc.stdout, + companion_procs=[enum_proc], + reraise_interrupt=True, + ) except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() - enum_proc.kill() + if enum_proc.stderr: + enum_proc.stderr.close() return + if enum_proc.stdout: + enum_proc.stdout.close() if enum_proc.returncode != 0: stderr_output = ( enum_proc.stderr.read().decode("utf-8", errors="replace").strip() @@ -2756,12 +2806,7 @@ def hcatGoodMeasure(hcatHashType, hcatHashFile): cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) cmd = _add_debug_mode_for_rules(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Good Measure", hash_file=hcatHashFile) hcatExtraCount = lineCount(hcatHashFile + ".out") - hcatHashCracked @@ -2789,11 +2834,12 @@ def hcatLMtoNT(): ] cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - hcatProcess.kill() + _run_hcat_cmd( + cmd, + attack_name="LM to NT (LM phase)", + hash_file=f"{hcatHashFile}.lm", + out_path=f"{hcatHashFile}.lm.cracked", + ) _write_delimited_field(f"{hcatHashFile}.lm.cracked", f"{hcatHashFile}.working", 2) converted = convert_hex("{hash_file}.working".format(hash_file=hcatHashFile)) @@ -2840,12 +2886,12 @@ def hcatLMtoNT(): cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) cmd = _add_debug_mode_for_rules(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print("Killing PID {0}...".format(str(hcatProcess.pid))) - hcatProcess.kill() + _run_hcat_cmd( + cmd, + attack_name="LM to NT (NT phase)", + hash_file=f"{hcatHashFile}.nt", + out_path=f"{hcatHashFile}.nt.out", + ) # toggle-lm-ntlm.rule by Didier Stevens https://blog.didierstevens.com/2016/07/16/tool-to-generate-hashcat-toggle-rules/ @@ -2882,11 +2928,7 @@ def hcatRecycle(hcatHashType, hcatHashFile, hcatNewPasswords): cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) cmd = _add_debug_mode_for_rules(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Recycle", hash_file=hcatHashFile) def hcatGenerateRules(hcatHashType, hcatHashFile, rule_count, wordlist): @@ -2922,12 +2964,7 @@ def hcatGenerateRules(hcatHashType, hcatHashFile, rule_count, wordlist): ] cmd.extend(shlex.split(hcatTuning)) _append_potfile_arg(cmd) - hcatProcess = subprocess.Popen(cmd) - try: - hcatProcess.wait() - except KeyboardInterrupt: - print(f"Killing PID {hcatProcess.pid}...") - hcatProcess.kill() + _run_hcat_cmd(cmd, attack_name="Random Rules", hash_file=hcatHashFile) finally: if os.path.exists(rules_path): os.unlink(rules_path) @@ -2990,7 +3027,9 @@ def cleanup(): if os.path.isfile(out_path): print(f"\nCracked passwords combined with original hashes in {out_path}") else: - print(f"\nNo cracked hashes to combine. Raw output (if any): {hcatHashFile}.out") + print( + f"\nNo cracked hashes to combine. Raw output (if any): {hcatHashFile}.out" + ) print("\nCleaning up temporary files...") if os.path.exists(hcatHashFile + ".masks"): os.remove(hcatHashFile + ".masks") @@ -3792,9 +3831,7 @@ def wordlist_filter_req_exclude(infile: str, outfile: str, mask: int) -> bool: return result.returncode == 0 -def wordlist_cutb( - infile: str, outfile: str, offset: int, length: int | None -) -> bool: +def wordlist_cutb(infile: str, outfile: str, offset: int, length: int | None) -> bool: """Extract a substring from each word starting at offset, optionally limited to length bytes.""" cutb_bin = os.path.join(hate_path, "hashcat-utils/bin/cutb.bin") cmd = [cutb_bin, str(offset)] @@ -3832,7 +3869,9 @@ def wordlist_gate(infile: str, outfile: str, mod: int, offset: int) -> bool: """Shard wordlist: keep every mod-th line starting at offset.""" gate_bin = os.path.join(hate_path, "hashcat-utils/bin/gate.bin") with open(infile, "rb") as fin, open(outfile, "wb") as fout: - result = subprocess.run([gate_bin, str(mod), str(offset)], stdin=fin, stdout=fout) + result = subprocess.run( + [gate_bin, str(mod), str(offset)], stdin=fin, stdout=fout + ) return result.returncode == 0 @@ -3850,7 +3889,9 @@ def rules_cleanup(infile: str, outfile: str) -> bool: def rules_optimize(infile: str, outfile: str) -> bool: """Optimize a rule file using rules_optimize.bin. Returns True on success.""" - optimize_path = os.path.join(hate_path, "hashcat-utils", "bin", "rules_optimize.bin") + optimize_path = os.path.join( + hate_path, "hashcat-utils", "bin", "rules_optimize.bin" + ) with open(infile, "rb") as fin, open(outfile, "wb") as fout: result = subprocess.run([optimize_path], stdin=fin, stdout=fout) return result.returncode == 0 @@ -3860,6 +3901,39 @@ def rule_tools_submenu(): return _attacks.rule_tools_submenu(_attack_ctx()) +def notifications_submenu(): + """Submenu for all Pushover notification controls (main-menu option 82). + + The inline ``interactive_menu`` import is not redundant with the + module-scope import at the top of this file: re-importing inside the + function re-reads ``hate_crack.menu.interactive_menu`` on every call, + which lets tests patch the real menu function via + ``monkeypatch.setattr(hate_crack.menu, "interactive_menu", ...)``. + Removing it breaks test isolation. + """ + from hate_crack.menu import interactive_menu + + while True: + settings = _notify.get_settings() + global_label = "ON" if settings.enabled else "OFF" + per_crack_label = "ON" if settings.per_crack_enabled else "OFF" + items = [ + ("1", f"Toggle Pushover Notifications [{global_label}]"), + ("2", f"Toggle Per-Crack Notifications [{per_crack_label}]"), + ("3", "Send Test Pushover Notification"), + ("99", "Back to Main Menu"), + ] + choice = interactive_menu(items, title="\nNotifications:") + if choice is None or choice == "99": + break + if choice == "1": + toggle_notifications() + elif choice == "2": + toggle_per_crack_notifications() + elif choice == "3": + test_pushover_notification() + + # convert hex words for recycling def convert_hex(working_file): processed_words = [] @@ -4067,6 +4141,82 @@ def quit_hc(): sys.exit(0) +def toggle_notifications(): + """Global on/off toggle for Pushover notifications. + + Flips ``notify_enabled`` in the active settings and persists to + ``config.json``. Prints the new state so the user has immediate + confirmation even though the menu label will also refresh on the + next render. + """ + new_state = _notify.toggle_enabled() + label = "ON" if new_state else "OFF" + print(f"\nPushover notifications are now {label}.") + if new_state: + settings = _notify.get_settings() + if not settings.pushover_token or not settings.pushover_user: + print( + "[!] notify_pushover_token / notify_pushover_user are empty in " + "config.json — notifications will silently no-op until set." + ) + + +def toggle_per_crack_notifications(): + """Runtime toggle for ``notify_per_crack_enabled`` with a UI-level guard. + + Per-crack notifications require global notifications to be ON in order + to fire (see ``notify.start_tailer``). Turning per-crack ON while the + global switch is OFF is silently ineffective, which surprises users — + so we refuse the transition and point them at the global toggle. + + Turning per-crack OFF is always allowed, regardless of the global + state, so users can clean up an inconsistent config without friction. + """ + settings = _notify.get_settings() + if not settings.per_crack_enabled and not settings.enabled: + print( + "\n[!] Global Pushover notifications are OFF. Enable option 1 " + "(Toggle Pushover Notifications) first." + ) + return + new_state = _notify.toggle_per_crack_enabled() + label = "ON" if new_state else "OFF" + print(f"\nPer-crack notifications are now {label}.") + + +def test_pushover_notification(): + """Send a canned test notification so the user can verify Pushover works. + + Ignores the global ``notify_enabled`` toggle on purpose: the point of the + test is to confirm the wire is live, independent of whether attacks are + currently wired to notify. When the global toggle is OFF we still send + but print a note so the user is not surprised later. + """ + settings = _notify.get_settings() + token = settings.pushover_token + user = settings.pushover_user + if not token or not user: + print( + "\n[!] Pushover credentials missing. Set notify_pushover_token " + "and notify_pushover_user in config.json." + ) + return + + if not settings.enabled: + print("\n(notifications are globally OFF, but sending test anyway)") + + title = "hate_crack: test notification" + message = ( + "This is a test notification from hate_crack. " + "If you see this, Pushover is wired up correctly." + ) + ok = _notify._send_pushover(token, user, title, message) + if ok: + print("[+] Test Pushover notification sent. Check your device.") + else: + print("[!] Test Pushover notification failed. See log output for details.") + + def get_main_menu_items(): """Return ordered (key, label) pairs for the main menu.""" items = [ @@ -4091,6 +4241,7 @@ def get_main_menu_items(): ("22", "Combipow Passphrase Attack"), ("80", "Wordlist Tools"), ("81", "Rule File Tools"), + ("82", "Notifications"), ("90", "Download rules from Hashmob.net"), ("91", "Analyze Hashcat Rules"), ("92", "Download wordlists from Hashmob.net"), @@ -4134,6 +4285,7 @@ def get_main_menu_options(): "22": combipow_crack, "80": wordlist_tools_submenu, "81": rule_tools_submenu, + "82": notifications_submenu, "90": lambda: download_hashmob_rules(rules_dir=rulesDirectory), "91": analyze_rules, "92": download_hashmob_wordlists, diff --git a/hate_crack/notify/__init__.py b/hate_crack/notify/__init__.py new file mode 100644 index 0000000..ea21ad3 --- /dev/null +++ b/hate_crack/notify/__init__.py @@ -0,0 +1,311 @@ +"""Public notification API for hate_crack. + +Overview +======== + +This package wires per-attack and per-crack notifications into the core +hashcat runner. The design is intentionally small and functional: a single +module-level ``_settings`` object, a handful of helper functions, and one +``CrackTailer`` thread class where polling state genuinely wants OO. + +Wiring +====== + +At startup ``main.py`` calls :func:`init` with the resolved config path and +parsed config dict. After that, the rest of the codebase interacts with +this package via: + +- :func:`prompt_notify_for_attack` -- called by attacks.py before an attack + starts; asks the user ``[y/n/always]`` and stashes per-run consent. +- :func:`start_tailer` / :func:`stop_tailer` -- called by the hashcat + command wrapper to spin a background watcher on ``{hashfile}.out``. +- :func:`notify_job_done` -- called by the hashcat command wrapper after + the subprocess exits, fires one summary notification. +- :func:`suppressed_notifications` -- context manager for orchestrator + attacks that chain many primitives; collapses all nested notifications + so the orchestrator can fire a single aggregate at the end. + +Adding a new backend +==================== + +The single concrete HTTP call lives in :mod:`hate_crack.notify.pushover`. +To add Slack/webhooks, write a sibling ``_send_slack()`` there (or in a +new module) and dispatch from :func:`notify_job_done` / +:func:`notify_crack`. No framework, no ABC — one function per transport. +""" + +from __future__ import annotations + +import logging +from typing import Callable + +from hate_crack.notify._suppress import ( + is_suppressed, + suppressed_notifications, +) +from hate_crack.notify.pushover import _send_pushover +from hate_crack.notify.settings import ( + NotifySettings, + add_to_allowlist, + load_settings, + save_enabled, + save_per_crack_enabled, +) +from hate_crack.notify.tailer import ( + CrackTailer, + extract_username_from_out_line, +) + +logger = logging.getLogger(__name__) + + +# Module-level runtime state. Treated as a singleton because the CLI is a +# single-process tool with a single user; no need to pass a context object +# through every attack signature. +_settings: NotifySettings | None = None +_config_path: str | None = None + +# Per-run consent cache: attack_name -> bool. Populated by +# ``prompt_notify_for_attack`` and consulted by ``notify_job_done`` so we +# don't need to re-prompt mid-chain. +_run_consent: dict[str, bool] = {} + +# Input function indirection so tests can inject answers without pulling +# in a terminal. Swap via ``set_input_func``. +_input_func: Callable[[str], str] = input + + +__all__ = [ + "CrackTailer", + "NotifySettings", + "add_to_allowlist", + "clear_state_for_tests", + "extract_username_from_out_line", + "get_settings", + "init", + "is_suppressed", + "notify_crack", + "notify_job_done", + "prompt_notify_for_attack", + "set_input_func", + "start_tailer", + "stop_tailer", + "suppressed_notifications", + "toggle_enabled", + "toggle_per_crack_enabled", + "_send_pushover", +] + + +def init(config_path: str | None, config_parser: dict | None) -> None: + """Bootstrap the notify subsystem from the resolved config. + + Called once from ``main.py`` after its config-loading block. Safe to + call multiple times — the second call replaces settings but does not + reset per-run consent (the user may already have answered prompts). + """ + global _settings, _config_path + _config_path = config_path + _settings = load_settings(config_parser) + + +def get_settings() -> NotifySettings: + """Return the active settings, or fresh defaults if ``init`` never ran.""" + return _settings if _settings is not None else NotifySettings() + + +def set_input_func(func: Callable[[str], str]) -> None: + """Test hook: swap the ``input()`` used by :func:`prompt_notify_for_attack`.""" + global _input_func + _input_func = func + + +def clear_state_for_tests() -> None: + """Reset module state. Only used by the test suite.""" + global _settings, _config_path, _input_func + _settings = None + _config_path = None + _run_consent.clear() + _input_func = input + + +def toggle_enabled() -> bool: + """Flip ``notify_enabled``, persist to ``config.json``, return new state. + + If ``init`` was never called we still toggle an in-memory default — the + UI update must not crash even if the config file is unreachable. + """ + global _settings + if _settings is None: + _settings = NotifySettings() + _settings.enabled = not _settings.enabled + if _config_path: + try: + save_enabled(_config_path, _settings.enabled) + except OSError as exc: + logger.warning("Could not persist notify_enabled: %s", exc) + return _settings.enabled + + +def toggle_per_crack_enabled() -> bool: + """Flip ``notify_per_crack_enabled``, persist to ``config.json``, return new state. + + If ``init`` was never called we still toggle an in-memory default — the + UI update must not crash even if the config file is unreachable. + """ + global _settings + if _settings is None: + _settings = NotifySettings() + _settings.per_crack_enabled = not _settings.per_crack_enabled + if _config_path: + try: + save_per_crack_enabled(_config_path, _settings.per_crack_enabled) + except OSError as exc: + logger.warning("Could not persist notify_per_crack_enabled: %s", exc) + return _settings.per_crack_enabled + + +def _in_allowlist(attack_name: str) -> bool: + return attack_name in get_settings().attack_allowlist + + +def prompt_notify_for_attack(attack_name: str) -> bool: + """Ask the user whether this attack should fire a notification. + + Returns ``True`` if notifications should fire for this run. + + Flow: + + 1. Notifications disabled globally -> return False silently (no prompt). + 2. Attack already in allowlist -> return True (no prompt, auto-on). + 3. Otherwise -> prompt ``[y/n/always]``: + * ``y`` -> consent for this run only. + * ``n`` / "" -> no consent. + * ``always`` -> persist to allowlist and consent for this run. + + Per-run consent is stashed in ``_run_consent[attack_name]`` so the + hashcat wrapper can query it at job-done time without re-prompting. + """ + settings = get_settings() + if not settings.enabled: + _run_consent[attack_name] = False + return False + if _in_allowlist(attack_name): + _run_consent[attack_name] = True + return True + + try: + raw = _input_func( + f"\n[notify] Send Pushover notifications for '{attack_name}'? [y/N/always]: " + ) + except EOFError: + raw = "" + answer = (raw or "").strip().lower() + if answer == "always": + _run_consent[attack_name] = True + if _config_path: + try: + add_to_allowlist(_config_path, attack_name) + # Also update the in-memory settings so a later call in the + # same session sees the allowlist without re-reading config. + if attack_name not in settings.attack_allowlist: + settings.attack_allowlist.append(attack_name) + except OSError as exc: + logger.warning("Could not persist allowlist entry: %s", exc) + return True + if answer in ("y", "yes"): + _run_consent[attack_name] = True + return True + _run_consent[attack_name] = False + return False + + +def _should_fire(attack_name: str) -> bool: + if is_suppressed(): + return False + settings = get_settings() + if not settings.enabled: + return False + if _in_allowlist(attack_name): + return True + return _run_consent.get(attack_name, False) + + +def notify_job_done( + attack_name: str, + cracked_count: int, + hash_file: str | None = None, +) -> None: + """Fire a single "attack complete" notification. + + No-op when suppressed, disabled, or the user declined at the prompt. + """ + if not _should_fire(attack_name): + return + settings = get_settings() + title = f"hate_crack: {attack_name} complete" + if hash_file: + message = ( + f"Attack '{attack_name}' finished.\n" + f"Cracked so far: {cracked_count}\n" + f"Hash file: {hash_file}" + ) + else: + message = f"Attack '{attack_name}' finished.\nCracked so far: {cracked_count}" + _send_pushover(settings.pushover_token, settings.pushover_user, title, message) + + +def notify_crack(label: str, attack_name: str) -> None: + """Fire a per-crack notification (called from :class:`CrackTailer`).""" + if not _should_fire(attack_name): + return + settings = get_settings() + title = "hate_crack: new crack" + message = f"{label} cracked ({attack_name})" + _send_pushover(settings.pushover_token, settings.pushover_user, title, message) + + +def _notify_aggregate(count: int, attack_name: str) -> None: + """Aggregated "N accounts cracked" notification for burst-capped ticks.""" + if not _should_fire(attack_name): + return + settings = get_settings() + title = "hate_crack: crack burst" + message = f"{count} new accounts cracked ({attack_name})" + _send_pushover(settings.pushover_token, settings.pushover_user, title, message) + + +def start_tailer(out_path: str, attack_name: str) -> CrackTailer | None: + """Start a :class:`CrackTailer` if per-crack notifications are enabled. + + Returns the running tailer (so the caller can stop it later), or + ``None`` when suppression/disabled/disallowed mean we shouldn't tail. + """ + if is_suppressed(): + return None + settings = get_settings() + if not settings.enabled: + return None + if not settings.per_crack_enabled: + return None + if not _should_fire(attack_name): + return None + tailer = CrackTailer( + out_path=out_path, + attack_name=attack_name, + settings=settings, + notify_callback=notify_crack, + aggregate_callback=_notify_aggregate, + ) + tailer.start() + return tailer + + +def stop_tailer(tailer: CrackTailer | None) -> None: + """Stop a tailer started by :func:`start_tailer`. ``None`` is a no-op.""" + if tailer is None: + return + try: + tailer.stop() + except Exception as exc: + logger.warning("CrackTailer.stop() failed: %s", exc) diff --git a/hate_crack/notify/_suppress.py b/hate_crack/notify/_suppress.py new file mode 100644 index 0000000..d70a40a --- /dev/null +++ b/hate_crack/notify/_suppress.py @@ -0,0 +1,38 @@ +"""Thread-local notification suppression. + +Multi-phase orchestrators (e.g. ``extensive_crack``) chain many hashcat +invocations. Without suppression every primitive would fire its own +"job complete" notification — loud, useless, and easy to rate-limit into +the ground. Callers wrap the chain in :func:`suppressed_notifications` +and then fire exactly one aggregate notification at the end. + +Suppression is thread-local so a tailer thread in a different context +can't accidentally silence itself by observing another thread's flag. +""" + +from __future__ import annotations + +import contextlib +import threading + +_suppressed = threading.local() + + +def is_suppressed() -> bool: + """Return True if the calling thread is inside a suppression context.""" + return getattr(_suppressed, "active", False) + + +@contextlib.contextmanager +def suppressed_notifications(): + """Context manager that suppresses notify_* calls on the current thread. + + Nests correctly: the previous state is restored on exit so an outer + ``with`` block still reflects "suppressed" after an inner one exits. + """ + prev = getattr(_suppressed, "active", False) + _suppressed.active = True + try: + yield + finally: + _suppressed.active = prev diff --git a/hate_crack/notify/pushover.py b/hate_crack/notify/pushover.py new file mode 100644 index 0000000..e0b3e74 --- /dev/null +++ b/hate_crack/notify/pushover.py @@ -0,0 +1,69 @@ +"""Pushover HTTP backend. + +Single concrete entry point: :func:`_send_pushover`. Exposed as a module-level +function (not a class) so that wiring in a Slack or generic webhook backend +later is a matter of writing a sibling ``_send_slack()`` alongside this one. + +Design invariants: + +- The call is a best-effort side effect. A crashed attack is strictly worse + than a missed notification, so every exception below the ``requests`` layer + is caught and swallowed (returning ``False``). +- Missing ``requests`` is treated the same as any other transport error — + we log a warning once and return ``False`` without re-raising. +- Payload contains *only* title + message + credentials. No plaintexts, no + hash data. The tailer guarantees this at call sites; this function is the + final safety barrier if a caller passes bad data (we still forward whatever + string it gives us, but we never introspect outside the message). +""" + +from __future__ import annotations + +import logging + +logger = logging.getLogger(__name__) + +try: + import requests # type: ignore[import-untyped] +except ImportError: # pragma: no cover - requests is a hard dep of the package + requests = None # type: ignore[assignment] + + +PUSHOVER_URL = "https://api.pushover.net/1/messages.json" + + +def _send_pushover(token: str, user: str, title: str, message: str) -> bool: + """POST a notification to Pushover. + + Returns ``True`` on HTTP 200, ``False`` on any other outcome. Never + raises — network errors, missing credentials, and missing ``requests`` + are all funneled into a ``False`` return so the caller can treat the + notification as a fire-and-forget side effect. + """ + if not token or not user: + logger.debug("Pushover not configured: missing token or user") + return False + + if requests is None: + logger.warning("Pushover requested but 'requests' is not importable") + return False + + payload = { + "token": token, + "user": user, + "title": title, + "message": message, + } + try: + response = requests.post(PUSHOVER_URL, data=payload, timeout=10) + except Exception as exc: + # requests.RequestException covers most, but a plugin/mock may raise + # a bare Exception — we must never let this escape. + logger.warning("Pushover send failed: %s", exc) + return False + + status = getattr(response, "status_code", None) + if status == 200: + return True + logger.warning("Pushover send returned HTTP %s", status) + return False diff --git a/hate_crack/notify/settings.py b/hate_crack/notify/settings.py new file mode 100644 index 0000000..d5f7fe1 --- /dev/null +++ b/hate_crack/notify/settings.py @@ -0,0 +1,179 @@ +"""Notification settings: dataclass + atomic config persistence. + +Settings live in the same ``config.json`` that drives the rest of hate_crack. +This module isolates (a) the typed shape of notification config and +(b) the read-modify-write persistence primitives used by the runtime toggle +and the ``[yes/no/always]`` per-attack prompt. + +Persistence follows the same pattern as ``main.py`` (``json.load`` -> +mutate -> ``json.dump(..., indent=2)`` via a temp file and ``os.replace``) +so a crash mid-write cannot corrupt the config. +""" + +from __future__ import annotations + +import json +import os +import tempfile +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class NotifySettings: + """Typed view of the ``notify_*`` keys from ``config.json``. + + Defaults mirror ``config.json.example`` so freshly-loaded configs and + in-memory fallbacks agree. + """ + + enabled: bool = False + pushover_token: str = "" + pushover_user: str = "" + per_crack_enabled: bool = False + attack_allowlist: list[str] = field(default_factory=list) + suppress_in_orchestrators: bool = True + max_cracks_per_burst: int = 5 + poll_interval_seconds: float = 5.0 + + +def _coerce_bool(value: Any, default: bool) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return bool(value) + if isinstance(value, str): + return value.strip().lower() in ("1", "true", "yes", "on") + return default + + +def _coerce_int(value: Any, default: int) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _coerce_float(value: Any, default: float) -> float: + try: + return float(value) + except (TypeError, ValueError): + return default + + +def _coerce_str(value: Any, default: str) -> str: + if value is None: + return default + return str(value) + + +def _coerce_list(value: Any) -> list[str]: + if isinstance(value, list): + return [str(v) for v in value] + return [] + + +def load_settings(config_parser: dict | None) -> NotifySettings: + """Build a ``NotifySettings`` from a parsed config dict. + + Unknown / missing / badly-typed keys fall back to dataclass defaults + so the runtime always has a valid settings object, even when the + config was written by an older hate_crack install. + """ + cfg = config_parser or {} + defaults = NotifySettings() + return NotifySettings( + enabled=_coerce_bool(cfg.get("notify_enabled"), defaults.enabled), + pushover_token=_coerce_str( + cfg.get("notify_pushover_token"), defaults.pushover_token + ), + pushover_user=_coerce_str( + cfg.get("notify_pushover_user"), defaults.pushover_user + ), + per_crack_enabled=_coerce_bool( + cfg.get("notify_per_crack_enabled"), defaults.per_crack_enabled + ), + attack_allowlist=_coerce_list(cfg.get("notify_attack_allowlist")), + suppress_in_orchestrators=_coerce_bool( + cfg.get("notify_suppress_in_orchestrators"), + defaults.suppress_in_orchestrators, + ), + max_cracks_per_burst=_coerce_int( + cfg.get("notify_max_cracks_per_burst"), defaults.max_cracks_per_burst + ), + poll_interval_seconds=_coerce_float( + cfg.get("notify_poll_interval_seconds"), defaults.poll_interval_seconds + ), + ) + + +def _atomic_rewrite(config_path: str, mutator) -> None: + """Read config_path, apply ``mutator(dict)`` in place, write atomically. + + - Missing file is treated as empty dict (mutator runs on ``{}``). + - Invalid JSON is silently replaced with the mutator's output; we do not + want to block a notification toggle on a pre-existing bad config. + - Write goes to a temp file in the same directory and is swapped in + via ``os.replace`` so readers never see a half-written file. + """ + data: dict = {} + if os.path.isfile(config_path): + try: + with open(config_path) as f: + loaded = json.load(f) + if isinstance(loaded, dict): + data = loaded + except (OSError, json.JSONDecodeError): + data = {} + mutator(data) + directory = os.path.dirname(os.path.abspath(config_path)) or "." + os.makedirs(directory, exist_ok=True) + fd, tmp_path = tempfile.mkstemp(prefix=".config-", suffix=".json", dir=directory) + try: + with os.fdopen(fd, "w") as tmp: + json.dump(data, tmp, indent=2) + os.replace(tmp_path, config_path) + except Exception: + # Best-effort cleanup of stale temp file on failure. + try: + os.unlink(tmp_path) + except OSError: + pass + raise + + +def save_enabled(config_path: str, enabled: bool) -> None: + """Persist ``notify_enabled`` without disturbing other config keys.""" + + def _apply(data: dict) -> None: + data["notify_enabled"] = bool(enabled) + + _atomic_rewrite(config_path, _apply) + + +def save_per_crack_enabled(config_path: str, enabled: bool) -> None: + """Persist ``notify_per_crack_enabled`` without disturbing other config keys.""" + + def _apply(data: dict) -> None: + data["notify_per_crack_enabled"] = bool(enabled) + + _atomic_rewrite(config_path, _apply) + + +def add_to_allowlist(config_path: str, attack_name: str) -> None: + """Append ``attack_name`` to ``notify_attack_allowlist`` if absent. + + Idempotent: already-present entries are a no-op. + """ + if not attack_name: + return + + def _apply(data: dict) -> None: + current = data.get("notify_attack_allowlist") + if not isinstance(current, list): + current = [] + if attack_name not in current: + current.append(attack_name) + data["notify_attack_allowlist"] = current + + _atomic_rewrite(config_path, _apply) diff --git a/hate_crack/notify/tailer.py b/hate_crack/notify/tailer.py new file mode 100644 index 0000000..0808eb1 --- /dev/null +++ b/hate_crack/notify/tailer.py @@ -0,0 +1,208 @@ +"""Background tailer that watches ``{hashfile}.out`` for new cracks. + +Design rationale: + +- Polling (not inotify/fsevents) keeps the implementation portable across + macOS/Linux without adding a dependency. +- We seek to EOF on start: the hashfile may already contain cracks from a + previous run, and re-notifying those would be both wrong and spammy. +- Burst cap: if a single poll tick yields more cracks than the user's + configured threshold, we collapse them into one aggregated notification + instead of N pings. Cracks tend to arrive in rule-file bursts, so this + knob matters. +- The thread is a daemon so a hung tailer can't block process exit. +- ``stop()`` joins with a timeout to guarantee forward progress even if the + poll loop is stuck on a slow filesystem. +""" + +from __future__ import annotations + +import logging +import os +import threading +from typing import Callable + +from hate_crack.notify.settings import NotifySettings + +logger = logging.getLogger(__name__) + + +def extract_username_from_out_line(line: str) -> str | None: + """Extract a username from a single hashcat ``.out`` line, if present. + + We support the four output layouts hate_crack can produce upstream: + + ======================= =========================================== ============ + Input shape Example Returns + ======================= =========================================== ============ + bare hash ``5f4dcc3b5aa:plaintext`` ``None`` + user:hash:plain ``alice:5f4dcc:plaintext`` ``"alice"`` + pwdump ``alice:1001:aad3b:31d6:::plaintext`` ``"alice"`` + NetNTLMv2 ``alice::DOMAIN:challenge:response:plain`` ``"alice"`` + ======================= =========================================== ============ + + The function deliberately NEVER returns the plaintext — that would + defeat the privacy guarantee documented on the notify public API. + """ + if not line: + return None + stripped = line.rstrip("\r\n") + if not stripped: + return None + + parts = stripped.split(":") + if len(parts) < 2: + return None + + # Bare hash (hash:plain): exactly two colons-separated fields and the + # first field is a plausible hash — any non-empty first field that is + # not a pure hex string may still be a username, so we fall through. + if len(parts) == 2: + return None + + first = parts[0] + if not first: + # Lines that start with ':' aren't in any format we care about. + return None + + # pwdump: user:RID:LM:NT:::plain -> field[1] is numeric RID + if len(parts) >= 7 and parts[1].isdigit(): + return first + + # NetNTLMv2: user::DOMAIN:... -> field[1] is empty + if len(parts) >= 5 and parts[1] == "": + return first + + # user:hash:plain style — assume first field is username. + # We only claim this when there's at least 3 fields so we don't + # mis-label a bare ``hash:plain`` as ``username:plain``. + if len(parts) >= 3: + return first + + return None + + +class CrackTailer(threading.Thread): + """Polls ``out_path`` for new cracked lines and dispatches notifications. + + The tailer is created and started by :func:`hate_crack.notify.start_tailer` + and stopped by :func:`hate_crack.notify.stop_tailer`. External callers + normally don't touch this class directly; it's public for tests. + """ + + daemon = True + + def __init__( + self, + out_path: str, + attack_name: str, + settings: NotifySettings, + notify_callback: Callable[[str, str], None], + aggregate_callback: Callable[[int, str], None], + *, + username_extractor: Callable[[str], str | None] | None = None, + ) -> None: + super().__init__(name=f"CrackTailer[{attack_name}]", daemon=True) + self.out_path = out_path + self.attack_name = attack_name + self.settings = settings + self._notify_crack = notify_callback + self._notify_aggregate = aggregate_callback + self._extract_username = username_extractor or extract_username_from_out_line + self._stop = threading.Event() + # Leftover bytes from the previous read that didn't end in \n. + self._buffer = b"" + # File position to read from; set on first successful open. + self._file_pos: int | None = None + + def stop(self) -> None: + """Signal the thread to exit and join with a bounded timeout. + + Safe to call on a tailer that was never started; in that case the + signal is still set (so a later ``start()`` would exit immediately) + and we skip the join that would otherwise raise. + """ + self._stop.set() + if self._started.is_set(): + self.join(timeout=10.0) + + def run(self) -> None: # pragma: no cover - thread entry, exercised via tests + try: + self._seek_to_eof() + # First tick of ``wait()`` blocks for the full interval; this is + # deliberate so we don't hammer the filesystem immediately after + # starting if hashcat hasn't written anything yet. + while not self._stop.wait(self.settings.poll_interval_seconds): + try: + self._poll_once() + except Exception as exc: + logger.warning("CrackTailer poll failed: %s", exc) + except Exception as exc: # pragma: no cover - defensive + logger.warning("CrackTailer crashed: %s", exc) + + def _seek_to_eof(self) -> None: + """Remember the current EOF so existing cracks aren't re-notified.""" + if os.path.isfile(self.out_path): + try: + self._file_pos = os.path.getsize(self.out_path) + except OSError: + self._file_pos = 0 + else: + self._file_pos = 0 + + def _poll_once(self) -> None: + if not os.path.isfile(self.out_path): + return + try: + size = os.path.getsize(self.out_path) + except OSError: + return + + if self._file_pos is None: + self._file_pos = size + return + + if size < self._file_pos: + # File was truncated / rewritten — reset rather than read garbage. + self._file_pos = 0 + self._buffer = b"" + + if size == self._file_pos: + return + + new_lines = self._read_new_lines(size) + if not new_lines: + return + + count = len(new_lines) + if count > self.settings.max_cracks_per_burst: + self._notify_aggregate(count, self.attack_name) + return + + for line_bytes in new_lines: + try: + line = line_bytes.decode("utf-8", errors="replace") + except Exception: + continue + label = self._extract_username(line) or self.attack_name + self._notify_crack(label, self.attack_name) + + def _read_new_lines(self, size: int) -> list[bytes]: + """Read from ``self._file_pos`` up to ``size`` and return full lines. + + Incomplete trailing bytes are buffered for the next tick. + """ + try: + with open(self.out_path, "rb") as f: + f.seek(self._file_pos) + data = f.read(size - self._file_pos) + except OSError: + return [] + self._file_pos = size + + combined = self._buffer + data + lines = combined.split(b"\n") + # The last element is either empty (buffer ended on \n) or a partial + # line — stash it for next poll either way. + self._buffer = lines.pop() if lines else b"" + return [line for line in lines if line] diff --git a/tests/conftest.py b/tests/conftest.py index 33c4af9..22eb2bc 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -20,3 +20,25 @@ def load_hate_crack_module(monkeypatch): @pytest.fixture def hc_module(monkeypatch): return load_hate_crack_module(monkeypatch) + + +@pytest.fixture(autouse=True) +def _isolate_notify_state(): + """Reset notify module state between tests. + + ``hate_crack.main`` calls ``notify.init()`` at import time with whatever + ``config.json`` is resolved from the user's environment (e.g. + ``~/.hate_crack/config.json``). If that config has + ``notify_enabled: true``, the per-attack prompt in ``attacks.py`` fires + ``input()`` during tests and blows up capture. Forcing the notify + package back to its disabled-by-default state before every test keeps + the suite hermetic regardless of the developer's local config. + """ + try: + from hate_crack import notify + except ImportError: + yield + return + notify.clear_state_for_tests() + yield + notify.clear_state_for_tests() diff --git a/tests/test_notifications_submenu.py b/tests/test_notifications_submenu.py new file mode 100644 index 0000000..e317239 --- /dev/null +++ b/tests/test_notifications_submenu.py @@ -0,0 +1,136 @@ +"""Unit tests for the Notifications submenu dispatcher (main-menu option 82). + +Patching note: ``notifications_submenu`` is defined in ``hate_crack/main.py`` +and resolves ``toggle_notifications`` / ``toggle_per_crack_notifications`` / +``test_pushover_notification`` against ``hate_crack.main``'s own globals. +We therefore patch that module directly — patching the ``hate_crack.py`` +proxy would have no effect on the submenu's internal dispatch. +""" + +import hate_crack.main as _main_mod +import hate_crack.menu as _menu_mod +from hate_crack.notify.settings import NotifySettings + + +def _stub_action_handlers(monkeypatch, calls): + monkeypatch.setattr( + _main_mod, "toggle_notifications", lambda: calls.append("toggle") + ) + monkeypatch.setattr( + _main_mod, + "toggle_per_crack_notifications", + lambda: calls.append("toggle_pc"), + ) + monkeypatch.setattr( + _main_mod, + "test_pushover_notification", + lambda: calls.append("test"), + ) + + +def _queue_menu_choices(monkeypatch, choices): + """Queue ``choices`` as sequential return values from ``interactive_menu``. + + Always appends a final ``"99"`` so the loop exits even if the caller + forgot — this mirrors how the real user ends a submenu. + """ + iterator = iter(list(choices) + ["99"]) + + def _fake_menu(items, title=""): + return next(iterator) + + monkeypatch.setattr(_menu_mod, "interactive_menu", _fake_menu) + + +class TestNotificationsSubmenu: + def test_choice_1_dispatches_toggle_notifications(self, monkeypatch): + calls = [] + _stub_action_handlers(monkeypatch, calls) + _queue_menu_choices(monkeypatch, ["1"]) + _main_mod.notifications_submenu() + assert calls == ["toggle"] + + def test_choice_2_dispatches_toggle_per_crack(self, monkeypatch): + calls = [] + _stub_action_handlers(monkeypatch, calls) + _queue_menu_choices(monkeypatch, ["2"]) + _main_mod.notifications_submenu() + assert calls == ["toggle_pc"] + + def test_choice_3_dispatches_test_pushover(self, monkeypatch): + calls = [] + _stub_action_handlers(monkeypatch, calls) + _queue_menu_choices(monkeypatch, ["3"]) + _main_mod.notifications_submenu() + assert calls == ["test"] + + def test_choice_99_exits_without_dispatch(self, monkeypatch): + calls = [] + _stub_action_handlers(monkeypatch, calls) + + def _only_99(items, title=""): + return "99" + + monkeypatch.setattr(_menu_mod, "interactive_menu", _only_99) + _main_mod.notifications_submenu() + assert calls == [] + + def test_none_choice_exits_without_dispatch(self, monkeypatch): + calls = [] + _stub_action_handlers(monkeypatch, calls) + + def _returns_none(items, title=""): + return None + + monkeypatch.setattr(_menu_mod, "interactive_menu", _returns_none) + _main_mod.notifications_submenu() + assert calls == [] + + def test_submenu_labels_reflect_live_settings(self, monkeypatch): + captured_items = {} + + def _capture(items, title=""): + captured_items["items"] = items + captured_items["title"] = title + return "99" + + monkeypatch.setattr(_menu_mod, "interactive_menu", _capture) + monkeypatch.setattr( + _main_mod._notify, + "get_settings", + lambda: NotifySettings(enabled=True, per_crack_enabled=False), + ) + _main_mod.notifications_submenu() + labels = {k: v for k, v in captured_items["items"]} + assert "ON" in labels["1"] + assert "OFF" in labels["2"] + assert labels["3"] == "Send Test Pushover Notification" + assert labels["99"] == "Back to Main Menu" + assert "Notifications" in captured_items["title"] + + def test_labels_refresh_between_iterations(self, monkeypatch): + # Guards against a regression where items are built once outside + # the while-loop: labels would go stale after a toggle. + settings = NotifySettings(enabled=False, per_crack_enabled=False) + monkeypatch.setattr(_main_mod._notify, "get_settings", lambda: settings) + + def _flip_enabled(): + settings.enabled = not settings.enabled + + monkeypatch.setattr(_main_mod, "toggle_notifications", _flip_enabled) + monkeypatch.setattr(_main_mod, "toggle_per_crack_notifications", lambda: None) + monkeypatch.setattr(_main_mod, "test_pushover_notification", lambda: None) + + captured = [] + choices = iter(["1", "99"]) + + def _fake_menu(items, title=""): + captured.append(dict(items)) + return next(choices) + + monkeypatch.setattr(_menu_mod, "interactive_menu", _fake_menu) + _main_mod.notifications_submenu() + + assert len(captured) == 2 + assert "[OFF]" in captured[0]["1"] + assert "[ON]" in captured[1]["1"] diff --git a/tests/test_notify_integration.py b/tests/test_notify_integration.py new file mode 100644 index 0000000..b757851 --- /dev/null +++ b/tests/test_notify_integration.py @@ -0,0 +1,242 @@ +"""Integration-style tests for the notify public API.""" +import json +from pathlib import Path +from unittest.mock import patch + +import pytest + +from hate_crack import notify + + +@pytest.fixture(autouse=True) +def _reset_notify_state(): + notify.clear_state_for_tests() + yield + notify.clear_state_for_tests() + + +def _init_with(tmp_path: Path, **kwargs) -> Path: + """Create a config file and initialize notify against it.""" + cfg = { + "notify_enabled": True, + "notify_pushover_token": "tok", + "notify_pushover_user": "usr", + } + cfg.update(kwargs) + config_path = tmp_path / "config.json" + config_path.write_text(json.dumps(cfg)) + notify.init(str(config_path), cfg) + return config_path + + +class TestNotifyJobDone: + def test_no_op_when_disabled(self, tmp_path: Path) -> None: + _init_with(tmp_path, notify_enabled=False) + with patch("hate_crack.notify._send_pushover") as send: + notify.notify_job_done("Brute Force", 3) + send.assert_not_called() + + def test_fires_when_enabled_and_in_allowlist(self, tmp_path: Path) -> None: + _init_with( + tmp_path, + notify_enabled=True, + notify_attack_allowlist=["Brute Force"], + ) + with patch("hate_crack.notify._send_pushover") as send: + notify.notify_job_done("Brute Force", 7, hash_file="/tmp/h") + assert send.called + args = send.call_args.args + # (token, user, title, message) + assert args[0] == "tok" + assert args[1] == "usr" + assert "Brute Force" in args[2] + assert "7" in args[3] + + def test_no_op_when_enabled_but_attack_not_allowed(self, tmp_path: Path) -> None: + _init_with(tmp_path, notify_enabled=True) + # No allowlist, no consent -> should not fire. + with patch("hate_crack.notify._send_pushover") as send: + notify.notify_job_done("Brute Force", 1) + send.assert_not_called() + + def test_fires_after_per_run_consent(self, tmp_path: Path) -> None: + _init_with(tmp_path, notify_enabled=True) + notify.set_input_func(lambda _prompt: "y") + assert notify.prompt_notify_for_attack("Brute Force") is True + with patch("hate_crack.notify._send_pushover") as send: + notify.notify_job_done("Brute Force", 1) + assert send.called + + def test_suppression_silences_fire(self, tmp_path: Path) -> None: + _init_with( + tmp_path, + notify_enabled=True, + notify_attack_allowlist=["Brute Force"], + ) + with patch("hate_crack.notify._send_pushover") as send: + with notify.suppressed_notifications(): + notify.notify_job_done("Brute Force", 1) + send.assert_not_called() + + def test_does_not_leak_plaintext_in_message(self, tmp_path: Path) -> None: + _init_with( + tmp_path, + notify_enabled=True, + notify_attack_allowlist=["Brute Force"], + ) + # Message we pass to notify_job_done is count + attack_name + hash + # path only. Confirm nothing in the payload contains a plausible + # plaintext password token. + with patch("hate_crack.notify._send_pushover") as send: + notify.notify_job_done("Brute Force", 42, hash_file="/tmp/h.txt") + title, message = send.call_args.args[2], send.call_args.args[3] + for banned in ("plaintext", "password=", "secret"): + assert banned not in title + assert banned not in message + + +class TestNotifyCrack: + def test_fires_when_allowed(self, tmp_path: Path) -> None: + _init_with( + tmp_path, + notify_enabled=True, + notify_attack_allowlist=["Brute Force"], + ) + with patch("hate_crack.notify._send_pushover") as send: + notify.notify_crack("alice", "Brute Force") + assert send.called + + def test_no_op_when_suppressed(self, tmp_path: Path) -> None: + _init_with( + tmp_path, + notify_enabled=True, + notify_attack_allowlist=["Brute Force"], + ) + with patch("hate_crack.notify._send_pushover") as send: + with notify.suppressed_notifications(): + notify.notify_crack("alice", "Brute Force") + send.assert_not_called() + + +class TestPromptNotifyForAttack: + def test_no_prompt_when_globally_disabled(self, tmp_path: Path) -> None: + _init_with(tmp_path, notify_enabled=False) + calls: list[str] = [] + notify.set_input_func(lambda p: calls.append(p) or "y") + assert notify.prompt_notify_for_attack("Brute Force") is False + assert calls == [] + + def test_no_prompt_when_already_in_allowlist(self, tmp_path: Path) -> None: + _init_with( + tmp_path, + notify_enabled=True, + notify_attack_allowlist=["Brute Force"], + ) + calls: list[str] = [] + notify.set_input_func(lambda p: calls.append(p) or "n") + assert notify.prompt_notify_for_attack("Brute Force") is True + assert calls == [] + + def test_answer_yes(self, tmp_path: Path) -> None: + _init_with(tmp_path, notify_enabled=True) + notify.set_input_func(lambda _: "y") + assert notify.prompt_notify_for_attack("Brute Force") is True + + def test_answer_no(self, tmp_path: Path) -> None: + _init_with(tmp_path, notify_enabled=True) + notify.set_input_func(lambda _: "n") + assert notify.prompt_notify_for_attack("Brute Force") is False + + def test_answer_empty_defaults_to_no(self, tmp_path: Path) -> None: + _init_with(tmp_path, notify_enabled=True) + notify.set_input_func(lambda _: "") + assert notify.prompt_notify_for_attack("Brute Force") is False + + def test_answer_always_persists_to_allowlist(self, tmp_path: Path) -> None: + config_path = _init_with(tmp_path, notify_enabled=True) + notify.set_input_func(lambda _: "always") + assert notify.prompt_notify_for_attack("Brute Force") is True + data = json.loads(config_path.read_text()) + assert "Brute Force" in data.get("notify_attack_allowlist", []) + # Settings in memory also updated so we don't re-prompt next call. + settings = notify.get_settings() + assert "Brute Force" in settings.attack_allowlist + + def test_always_is_idempotent(self, tmp_path: Path) -> None: + config_path = _init_with(tmp_path, notify_enabled=True) + notify.set_input_func(lambda _: "always") + notify.prompt_notify_for_attack("Brute Force") + notify.prompt_notify_for_attack("Brute Force") + data = json.loads(config_path.read_text()) + assert data["notify_attack_allowlist"].count("Brute Force") == 1 + + +class TestToggleEnabled: + def test_toggle_flips_and_persists(self, tmp_path: Path) -> None: + config_path = _init_with(tmp_path, notify_enabled=False) + assert notify.get_settings().enabled is False + new_state = notify.toggle_enabled() + assert new_state is True + assert notify.get_settings().enabled is True + data = json.loads(config_path.read_text()) + assert data["notify_enabled"] is True + + # Flip back. + assert notify.toggle_enabled() is False + data = json.loads(config_path.read_text()) + assert data["notify_enabled"] is False + + def test_toggle_without_init_still_works(self) -> None: + notify.clear_state_for_tests() + # Call toggle before init ever ran — should just flip in-memory. + assert notify.toggle_enabled() is True + assert notify.get_settings().enabled is True + + +class TestStartStopTailer: + def test_start_tailer_noop_when_disabled(self, tmp_path: Path) -> None: + _init_with(tmp_path, notify_enabled=False, notify_per_crack_enabled=True) + t = notify.start_tailer(str(tmp_path / "h.out"), "Brute Force") + assert t is None + + def test_start_tailer_noop_when_per_crack_disabled(self, tmp_path: Path) -> None: + _init_with( + tmp_path, + notify_enabled=True, + notify_per_crack_enabled=False, + notify_attack_allowlist=["Brute Force"], + ) + t = notify.start_tailer(str(tmp_path / "h.out"), "Brute Force") + assert t is None + + def test_start_tailer_noop_when_suppressed(self, tmp_path: Path) -> None: + _init_with( + tmp_path, + notify_enabled=True, + notify_per_crack_enabled=True, + notify_attack_allowlist=["Brute Force"], + ) + with notify.suppressed_notifications(): + t = notify.start_tailer(str(tmp_path / "h.out"), "Brute Force") + assert t is None + + def test_start_tailer_when_enabled_and_consented(self, tmp_path: Path) -> None: + _init_with( + tmp_path, + notify_enabled=True, + notify_per_crack_enabled=True, + notify_attack_allowlist=["Brute Force"], + notify_poll_interval_seconds=0.05, + ) + out = tmp_path / "h.out" + out.write_text("") + t = notify.start_tailer(str(out), "Brute Force") + try: + assert t is not None + assert t.is_alive() + finally: + notify.stop_tailer(t) + + def test_stop_tailer_none_is_noop(self) -> None: + # Must not raise. + notify.stop_tailer(None) diff --git a/tests/test_notify_per_crack_toggle.py b/tests/test_notify_per_crack_toggle.py new file mode 100644 index 0000000..8a2c780 --- /dev/null +++ b/tests/test_notify_per_crack_toggle.py @@ -0,0 +1,135 @@ +"""Unit tests for the toggle_per_crack_enabled runtime toggle.""" + +import importlib.util +import json +from pathlib import Path + +from hate_crack import notify as _notify + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +_CLI_SPEC = importlib.util.spec_from_file_location( + "hate_crack_cli_percrack", PROJECT_ROOT / "hate_crack.py" +) +CLI_MODULE = importlib.util.module_from_spec(_CLI_SPEC) +_CLI_SPEC.loader.exec_module(CLI_MODULE) + + +def _init_with(tmp_path: Path, **overrides) -> Path: + """Seed a config file with defaults + overrides and init the notify module.""" + config_path = tmp_path / "config.json" + cfg = { + "notify_enabled": False, + "notify_per_crack_enabled": False, + "notify_pushover_token": "", + "notify_pushover_user": "", + } + cfg.update(overrides) + config_path.write_text(json.dumps(cfg)) + _notify.init(str(config_path), cfg) + return config_path + + +class TestTogglePerCrackEnabled: + def test_off_to_on_flips_and_persists(self, tmp_path: Path) -> None: + config_path = _init_with(tmp_path) + try: + new_state = _notify.toggle_per_crack_enabled() + assert new_state is True + assert _notify.get_settings().per_crack_enabled is True + data = json.loads(config_path.read_text()) + assert data["notify_per_crack_enabled"] is True + finally: + _notify.clear_state_for_tests() + + def test_on_to_off_flips_and_persists(self, tmp_path: Path) -> None: + config_path = _init_with(tmp_path, notify_per_crack_enabled=True) + try: + new_state = _notify.toggle_per_crack_enabled() + assert new_state is False + assert _notify.get_settings().per_crack_enabled is False + data = json.loads(config_path.read_text()) + assert data["notify_per_crack_enabled"] is False + finally: + _notify.clear_state_for_tests() + + def test_toggle_without_init_uses_defaults(self) -> None: + # Mirrors the behavior of toggle_enabled: must not crash when init + # was never called. The toggle flips an in-memory default; nothing + # is persisted because _config_path is None. + try: + _notify.clear_state_for_tests() + new_state = _notify.toggle_per_crack_enabled() + assert new_state is True + assert _notify.get_settings().per_crack_enabled is True + finally: + _notify.clear_state_for_tests() + + def test_does_not_touch_global_enabled(self, tmp_path: Path) -> None: + config_path = _init_with(tmp_path, notify_enabled=False) + try: + _notify.toggle_per_crack_enabled() + data = json.loads(config_path.read_text()) + # notify_enabled stays False; only per-crack flipped. + assert data["notify_enabled"] is False + assert data["notify_per_crack_enabled"] is True + finally: + _notify.clear_state_for_tests() + + +class TestTogglePerCrackNotificationsUI: + def _seed_settings(self, monkeypatch, *, enabled: bool, per_crack: bool): + from hate_crack.notify.settings import NotifySettings + + settings = NotifySettings(enabled=enabled, per_crack_enabled=per_crack) + monkeypatch.setattr(CLI_MODULE._notify, "get_settings", lambda: settings) + return settings + + def test_guard_refuses_on_when_global_off(self, monkeypatch, capsys): + self._seed_settings(monkeypatch, enabled=False, per_crack=False) + called = {"n": 0} + + def _fake_toggle() -> bool: + called["n"] += 1 + return True + + monkeypatch.setattr( + CLI_MODULE._notify, "toggle_per_crack_enabled", _fake_toggle + ) + CLI_MODULE.toggle_per_crack_notifications() + captured = capsys.readouterr().out + assert "Global Pushover notifications are OFF" in captured + assert called["n"] == 0 + + def test_flips_on_when_global_on(self, monkeypatch, capsys): + self._seed_settings(monkeypatch, enabled=True, per_crack=False) + called = {"n": 0} + + def _fake_toggle() -> bool: + called["n"] += 1 + return True + + monkeypatch.setattr( + CLI_MODULE._notify, "toggle_per_crack_enabled", _fake_toggle + ) + CLI_MODULE.toggle_per_crack_notifications() + captured = capsys.readouterr().out + assert "Per-crack notifications are now ON" in captured + assert called["n"] == 1 + + def test_off_to_off_is_allowed_even_if_global_off(self, monkeypatch, capsys): + # Turning OFF must always succeed, even with global OFF, so a user + # can clean up an inconsistent (per_crack=True, enabled=False) config. + self._seed_settings(monkeypatch, enabled=False, per_crack=True) + called = {"n": 0} + + def _fake_toggle() -> bool: + called["n"] += 1 + return False + + monkeypatch.setattr( + CLI_MODULE._notify, "toggle_per_crack_enabled", _fake_toggle + ) + CLI_MODULE.toggle_per_crack_notifications() + captured = capsys.readouterr().out + assert "Per-crack notifications are now OFF" in captured + assert called["n"] == 1 diff --git a/tests/test_notify_pushover.py b/tests/test_notify_pushover.py new file mode 100644 index 0000000..8b9f759 --- /dev/null +++ b/tests/test_notify_pushover.py @@ -0,0 +1,88 @@ +"""Unit tests for the Pushover HTTP backend.""" +from unittest.mock import MagicMock, patch + +from hate_crack.notify import pushover + + +def _mock_response(status_code: int = 200) -> MagicMock: + r = MagicMock() + r.status_code = status_code + return r + + +class TestSendPushoverSuccess: + def test_returns_true_on_http_200(self) -> None: + with patch.object(pushover, "requests") as mock_requests: + mock_requests.post.return_value = _mock_response(200) + ok = pushover._send_pushover("tok", "usr", "title", "msg") + assert ok is True + + def test_payload_has_no_plaintext(self) -> None: + with patch.object(pushover, "requests") as mock_requests: + mock_requests.post.return_value = _mock_response(200) + pushover._send_pushover("tok", "usr", "Crack complete", "alice cracked") + (url,), kwargs = mock_requests.post.call_args + assert url == pushover.PUSHOVER_URL + data = kwargs["data"] + assert set(data.keys()) == {"token", "user", "title", "message"} + assert data["token"] == "tok" + assert data["user"] == "usr" + # The whole payload is just title + message + creds. No 'password', + # 'hash', 'plaintext' or similar keys ever appear. + assert "password" not in data + assert "plaintext" not in data + assert "hash" not in data + + def test_timeout_is_passed(self) -> None: + with patch.object(pushover, "requests") as mock_requests: + mock_requests.post.return_value = _mock_response(200) + pushover._send_pushover("tok", "usr", "t", "m") + _, kwargs = mock_requests.post.call_args + assert kwargs["timeout"] == 10 + + +class TestSendPushoverFailureModes: + def test_missing_token_returns_false_without_calling_requests(self) -> None: + with patch.object(pushover, "requests") as mock_requests: + ok = pushover._send_pushover("", "usr", "t", "m") + assert ok is False + mock_requests.post.assert_not_called() + + def test_missing_user_returns_false(self) -> None: + with patch.object(pushover, "requests") as mock_requests: + ok = pushover._send_pushover("tok", "", "t", "m") + assert ok is False + mock_requests.post.assert_not_called() + + def test_network_error_returns_false(self) -> None: + with patch.object(pushover, "requests") as mock_requests: + mock_requests.post.side_effect = ConnectionError("refused") + ok = pushover._send_pushover("tok", "usr", "t", "m") + assert ok is False + + def test_generic_exception_returns_false(self) -> None: + with patch.object(pushover, "requests") as mock_requests: + mock_requests.post.side_effect = RuntimeError("boom") + ok = pushover._send_pushover("tok", "usr", "t", "m") + assert ok is False + + def test_non_200_response_returns_false(self) -> None: + with patch.object(pushover, "requests") as mock_requests: + mock_requests.post.return_value = _mock_response(500) + ok = pushover._send_pushover("tok", "usr", "t", "m") + assert ok is False + + def test_missing_requests_returns_false(self) -> None: + with patch.object(pushover, "requests", None): + ok = pushover._send_pushover("tok", "usr", "t", "m") + assert ok is False + + def test_never_raises(self) -> None: + # Even if requests.post returns something weird (no status_code), + # the function must not raise. + bad = MagicMock() + # Accessing .status_code returns a non-int Mock — it will not equal 200. + with patch.object(pushover, "requests") as mock_requests: + mock_requests.post.return_value = bad + ok = pushover._send_pushover("tok", "usr", "t", "m") + assert ok is False diff --git a/tests/test_notify_settings.py b/tests/test_notify_settings.py new file mode 100644 index 0000000..34d592b --- /dev/null +++ b/tests/test_notify_settings.py @@ -0,0 +1,198 @@ +"""Unit tests for hate_crack.notify.settings.""" + +import json +from pathlib import Path + +from hate_crack.notify.settings import ( + NotifySettings, + add_to_allowlist, + load_settings, + save_enabled, + save_per_crack_enabled, +) + + +class TestNotifySettingsDataclass: + def test_defaults(self) -> None: + s = NotifySettings() + assert s.enabled is False + assert s.pushover_token == "" + assert s.pushover_user == "" + assert s.per_crack_enabled is False + assert s.attack_allowlist == [] + assert s.suppress_in_orchestrators is True + assert s.max_cracks_per_burst == 5 + assert s.poll_interval_seconds == 5.0 + + def test_allowlist_default_is_fresh_per_instance(self) -> None: + # field(default_factory=list) must not share state. + a = NotifySettings() + b = NotifySettings() + a.attack_allowlist.append("Brute Force") + assert b.attack_allowlist == [] + + +class TestLoadSettings: + def test_load_from_empty_dict_returns_defaults(self) -> None: + s = load_settings({}) + assert s == NotifySettings() + + def test_load_from_none_returns_defaults(self) -> None: + assert load_settings(None) == NotifySettings() + + def test_load_full_dict(self) -> None: + s = load_settings( + { + "notify_enabled": True, + "notify_pushover_token": "tok", + "notify_pushover_user": "usr", + "notify_per_crack_enabled": True, + "notify_attack_allowlist": ["Brute Force", "Dictionary"], + "notify_suppress_in_orchestrators": False, + "notify_max_cracks_per_burst": 20, + "notify_poll_interval_seconds": 2.5, + } + ) + assert s.enabled is True + assert s.pushover_token == "tok" + assert s.pushover_user == "usr" + assert s.per_crack_enabled is True + assert s.attack_allowlist == ["Brute Force", "Dictionary"] + assert s.suppress_in_orchestrators is False + assert s.max_cracks_per_burst == 20 + assert s.poll_interval_seconds == 2.5 + + def test_load_tolerates_bad_types(self) -> None: + s = load_settings( + { + "notify_enabled": "true", + "notify_max_cracks_per_burst": "not-a-number", + "notify_poll_interval_seconds": "also-bad", + "notify_attack_allowlist": "not-a-list", + } + ) + # string "true" -> True + assert s.enabled is True + # bad ints fall back to defaults (5, 5.0) + assert s.max_cracks_per_burst == 5 + assert s.poll_interval_seconds == 5.0 + # non-list allowlist becomes empty list + assert s.attack_allowlist == [] + + +class TestSaveEnabled: + def test_writes_new_config(self, tmp_path: Path) -> None: + config_path = tmp_path / "config.json" + save_enabled(str(config_path), True) + data = json.loads(config_path.read_text()) + assert data["notify_enabled"] is True + + def test_preserves_existing_keys(self, tmp_path: Path) -> None: + config_path = tmp_path / "config.json" + initial = { + "hcatBin": "hashcat", + "hashview_api_key": "secret", + "notify_enabled": False, + } + config_path.write_text(json.dumps(initial)) + save_enabled(str(config_path), True) + data = json.loads(config_path.read_text()) + assert data["hcatBin"] == "hashcat" + assert data["hashview_api_key"] == "secret" + assert data["notify_enabled"] is True + + def test_toggles_back_and_forth(self, tmp_path: Path) -> None: + config_path = tmp_path / "config.json" + save_enabled(str(config_path), True) + save_enabled(str(config_path), False) + data = json.loads(config_path.read_text()) + assert data["notify_enabled"] is False + + def test_invalid_existing_config_replaced(self, tmp_path: Path) -> None: + config_path = tmp_path / "config.json" + config_path.write_text("this is not json") + save_enabled(str(config_path), True) + data = json.loads(config_path.read_text()) + assert data == {"notify_enabled": True} + + def test_atomic_no_half_write(self, tmp_path: Path) -> None: + # A partial write should never leave the main file invalid. We + # check that after save_enabled, parsing always succeeds. + config_path = tmp_path / "config.json" + for flag in (True, False, True, False, True): + save_enabled(str(config_path), flag) + json.loads(config_path.read_text()) # must not raise + + +class TestAddToAllowlist: + def test_adds_to_empty_list(self, tmp_path: Path) -> None: + config_path = tmp_path / "config.json" + add_to_allowlist(str(config_path), "Brute Force") + data = json.loads(config_path.read_text()) + assert data["notify_attack_allowlist"] == ["Brute Force"] + + def test_idempotent(self, tmp_path: Path) -> None: + config_path = tmp_path / "config.json" + add_to_allowlist(str(config_path), "Brute Force") + add_to_allowlist(str(config_path), "Brute Force") + add_to_allowlist(str(config_path), "Brute Force") + data = json.loads(config_path.read_text()) + assert data["notify_attack_allowlist"] == ["Brute Force"] + + def test_preserves_other_entries(self, tmp_path: Path) -> None: + config_path = tmp_path / "config.json" + config_path.write_text( + json.dumps( + { + "hcatBin": "hashcat", + "notify_attack_allowlist": ["Existing"], + } + ) + ) + add_to_allowlist(str(config_path), "Brute Force") + data = json.loads(config_path.read_text()) + assert data["hcatBin"] == "hashcat" + assert data["notify_attack_allowlist"] == ["Existing", "Brute Force"] + + def test_empty_attack_name_is_noop(self, tmp_path: Path) -> None: + config_path = tmp_path / "config.json" + config_path.write_text(json.dumps({"notify_attack_allowlist": ["A"]})) + add_to_allowlist(str(config_path), "") + data = json.loads(config_path.read_text()) + assert data["notify_attack_allowlist"] == ["A"] + + def test_repairs_non_list_allowlist(self, tmp_path: Path) -> None: + config_path = tmp_path / "config.json" + config_path.write_text(json.dumps({"notify_attack_allowlist": "bogus"})) + add_to_allowlist(str(config_path), "Brute Force") + data = json.loads(config_path.read_text()) + assert data["notify_attack_allowlist"] == ["Brute Force"] + + +class TestSavePerCrackEnabled: + def test_writes_new_config(self, tmp_path: Path) -> None: + config_path = tmp_path / "config.json" + save_per_crack_enabled(str(config_path), True) + data = json.loads(config_path.read_text()) + assert data["notify_per_crack_enabled"] is True + + def test_preserves_existing_keys(self, tmp_path: Path) -> None: + config_path = tmp_path / "config.json" + initial = { + "hcatBin": "hashcat", + "notify_enabled": True, + "notify_per_crack_enabled": False, + } + config_path.write_text(json.dumps(initial)) + save_per_crack_enabled(str(config_path), True) + data = json.loads(config_path.read_text()) + assert data["hcatBin"] == "hashcat" + assert data["notify_enabled"] is True + assert data["notify_per_crack_enabled"] is True + + def test_toggles_back_and_forth(self, tmp_path: Path) -> None: + config_path = tmp_path / "config.json" + save_per_crack_enabled(str(config_path), True) + save_per_crack_enabled(str(config_path), False) + data = json.loads(config_path.read_text()) + assert data["notify_per_crack_enabled"] is False diff --git a/tests/test_notify_suppression.py b/tests/test_notify_suppression.py new file mode 100644 index 0000000..c57557f --- /dev/null +++ b/tests/test_notify_suppression.py @@ -0,0 +1,74 @@ +"""Unit tests for hate_crack.notify._suppress.""" +import threading +import time + +from hate_crack.notify._suppress import is_suppressed, suppressed_notifications + + +class TestSuppressionContextManager: + def test_default_is_not_suppressed(self) -> None: + assert is_suppressed() is False + + def test_inside_context_is_suppressed(self) -> None: + assert is_suppressed() is False + with suppressed_notifications(): + assert is_suppressed() is True + assert is_suppressed() is False + + def test_nested_restores_outer_state(self) -> None: + with suppressed_notifications(): + assert is_suppressed() is True + with suppressed_notifications(): + assert is_suppressed() is True + # Leaving inner context must still leave us suppressed. + assert is_suppressed() is True + assert is_suppressed() is False + + def test_exception_restores_state(self) -> None: + try: + with suppressed_notifications(): + raise RuntimeError("boom") + except RuntimeError: + pass + assert is_suppressed() is False + + +class TestSuppressionThreadLocal: + def test_other_thread_not_suppressed_by_us(self) -> None: + seen: list[bool] = [] + ready = threading.Event() + done = threading.Event() + + def worker() -> None: + # Wait until main thread has entered its suppression context, + # then sample our own state. + ready.wait(timeout=2.0) + seen.append(is_suppressed()) + done.set() + + t = threading.Thread(target=worker, daemon=True) + t.start() + with suppressed_notifications(): + assert is_suppressed() is True + ready.set() + done.wait(timeout=2.0) + t.join(timeout=2.0) + assert seen == [False], f"worker thread saw suppression state: {seen}" + + def test_worker_thread_suppression_does_not_leak(self) -> None: + worker_done = threading.Event() + + def worker() -> None: + with suppressed_notifications(): + assert is_suppressed() is True + # Hold long enough that the main thread can sample. + time.sleep(0.05) + worker_done.set() + + t = threading.Thread(target=worker, daemon=True) + t.start() + # Main thread should never observe the worker's suppression. + assert is_suppressed() is False + t.join(timeout=2.0) + assert worker_done.is_set() + assert is_suppressed() is False diff --git a/tests/test_notify_tailer.py b/tests/test_notify_tailer.py new file mode 100644 index 0000000..7404ed2 --- /dev/null +++ b/tests/test_notify_tailer.py @@ -0,0 +1,219 @@ +"""Unit tests for the CrackTailer polling thread and username extractor.""" +import time +from pathlib import Path +from unittest.mock import MagicMock + +from hate_crack.notify.settings import NotifySettings +from hate_crack.notify.tailer import CrackTailer, extract_username_from_out_line + + +def _settings(**overrides) -> NotifySettings: + defaults = { + "enabled": True, + "per_crack_enabled": True, + "max_cracks_per_burst": 5, + # Short poll so tests don't hang. + "poll_interval_seconds": 0.05, + } + defaults.update(overrides) + return NotifySettings(**defaults) + + +def _make_tailer(out_path: Path, **overrides): + notify = MagicMock(name="notify_callback") + aggregate = MagicMock(name="aggregate_callback") + settings = _settings(**overrides) + tailer = CrackTailer( + out_path=str(out_path), + attack_name="Brute Force", + settings=settings, + notify_callback=notify, + aggregate_callback=aggregate, + ) + return tailer, notify, aggregate + + +def _wait_until(predicate, timeout=2.0): + """Spin until predicate() or timeout; returns whether it passed.""" + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(0.02) + return predicate() + + +class TestExtractUsername: + def test_bare_hash_returns_none(self) -> None: + assert extract_username_from_out_line("5f4dcc3b5aa10b0be:secret") is None + + def test_user_hash_plain(self) -> None: + assert extract_username_from_out_line("alice:5f4dcc:secret") == "alice" + + def test_pwdump(self) -> None: + assert ( + extract_username_from_out_line("alice:1001:aad3b435:31d6cfe:::secret") + == "alice" + ) + + def test_netntlmv2(self) -> None: + line = "alice::DOMAIN:1122334455667788:resp:resp2:secret" + assert extract_username_from_out_line(line) == "alice" + + def test_empty_line(self) -> None: + assert extract_username_from_out_line("") is None + + def test_whitespace_only(self) -> None: + assert extract_username_from_out_line("\n") is None + + def test_line_with_trailing_newline(self) -> None: + assert extract_username_from_out_line("alice:hash:plain\n") == "alice" + + def test_does_not_leak_plaintext(self) -> None: + # No matter the format, the return value must never equal the + # plaintext password. Paranoid test; relies on us knowing where + # the plaintext sits in each format. + for line, plain in [ + ("bob:1000:aa:bb:::s3cret", "s3cret"), + ("bob:hash:s3cret", "s3cret"), + ("bob::DOM:chal:resp:resp2:s3cret", "s3cret"), + ]: + assert extract_username_from_out_line(line) != plain + + +class TestCrackTailerStart: + def test_daemon_is_true(self, tmp_path: Path) -> None: + out = tmp_path / "hashes.out" + out.write_text("") + tailer, _, _ = _make_tailer(out) + assert tailer.daemon is True + + def test_seeks_to_eof_on_start(self, tmp_path: Path) -> None: + out = tmp_path / "hashes.out" + out.write_text("alice:hash:plain\nbob:hash:plain\n") + tailer, notify, aggregate = _make_tailer(out) + tailer.start() + try: + # Allow at least one poll interval; no new lines were added, so + # no notifications should fire. + time.sleep(0.2) + assert notify.call_count == 0 + assert aggregate.call_count == 0 + finally: + tailer.stop() + + def test_new_lines_fire_notify(self, tmp_path: Path) -> None: + out = tmp_path / "hashes.out" + out.write_text("") + tailer, notify, aggregate = _make_tailer(out) + tailer.start() + try: + with open(out, "a") as f: + f.write("alice:hash:plain\n") + f.write("bob:hash:plain\n") + assert _wait_until(lambda: notify.call_count >= 2) + assert aggregate.call_count == 0 + labels = [call.args[0] for call in notify.call_args_list] + assert "alice" in labels + assert "bob" in labels + finally: + tailer.stop() + + def test_no_username_falls_back_to_attack_name(self, tmp_path: Path) -> None: + out = tmp_path / "hashes.out" + out.write_text("") + tailer, notify, aggregate = _make_tailer(out) + tailer.start() + try: + with open(out, "a") as f: + # Bare-hash format -> extractor returns None -> fallback. + f.write("5f4dcc3b5:plain\n") + assert _wait_until(lambda: notify.call_count >= 1) + assert notify.call_args.args[0] == "Brute Force" + finally: + tailer.stop() + + +class TestCrackTailerBurstCap: + def test_burst_cap_fires_aggregate(self, tmp_path: Path) -> None: + out = tmp_path / "hashes.out" + out.write_text("") + tailer, notify, aggregate = _make_tailer(out, max_cracks_per_burst=3) + tailer.start() + try: + # Write 10 lines in one shot; a single poll tick must see them + # all and collapse into one aggregate call. + with open(out, "a") as f: + for i in range(10): + f.write(f"user{i}:hash:plain\n") + assert _wait_until(lambda: aggregate.call_count >= 1) + # Per-crack path must NOT have fired for this burst. + assert notify.call_count == 0 + args = aggregate.call_args.args + assert args[0] == 10 + assert args[1] == "Brute Force" + finally: + tailer.stop() + + def test_under_cap_fires_per_crack(self, tmp_path: Path) -> None: + out = tmp_path / "hashes.out" + out.write_text("") + tailer, notify, aggregate = _make_tailer(out, max_cracks_per_burst=10) + tailer.start() + try: + with open(out, "a") as f: + for i in range(3): + f.write(f"user{i}:hash:plain\n") + assert _wait_until(lambda: notify.call_count >= 3) + assert aggregate.call_count == 0 + finally: + tailer.stop() + + +class TestCrackTailerStop: + def test_stop_joins_within_timeout(self, tmp_path: Path) -> None: + out = tmp_path / "hashes.out" + out.write_text("") + tailer, _, _ = _make_tailer(out) + tailer.start() + start = time.time() + tailer.stop() + elapsed = time.time() - start + assert not tailer.is_alive() + assert elapsed < 5.0 + + def test_stop_is_safe_without_start(self, tmp_path: Path) -> None: + out = tmp_path / "hashes.out" + out.write_text("") + tailer, _, _ = _make_tailer(out) + # Calling stop on a never-started thread should not raise. + tailer.stop() + + +class TestCrackTailerFileHandling: + def test_missing_file_then_appearing(self, tmp_path: Path) -> None: + out = tmp_path / "missing.out" + tailer, notify, _ = _make_tailer(out) + tailer.start() + try: + time.sleep(0.15) + out.write_text("alice:hash:plain\n") + assert _wait_until(lambda: notify.call_count >= 1) + finally: + tailer.stop() + + def test_truncation_resets_position(self, tmp_path: Path) -> None: + out = tmp_path / "hashes.out" + out.write_text("alice:hash:plain\nbob:hash:plain\n") + tailer, notify, _ = _make_tailer(out) + tailer.start() + try: + time.sleep(0.15) + # Truncate the file and write a fresh line; tailer should reset + # its file position and see the new line. + out.write_text("charlie:hash:plain\n") + assert _wait_until(lambda: notify.call_count >= 1) + labels = [call.args[0] for call in notify.call_args_list] + assert "charlie" in labels + finally: + tailer.stop() diff --git a/tests/test_run_hcat_cmd.py b/tests/test_run_hcat_cmd.py new file mode 100644 index 0000000..0cd2474 --- /dev/null +++ b/tests/test_run_hcat_cmd.py @@ -0,0 +1,218 @@ +"""Tests for the ``_run_hcat_cmd`` subprocess/notify wrapper in main.py.""" +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.fixture +def main_module(hc_module): + return hc_module._main + + +def _make_mock_proc(wait_side_effect=None): + proc = MagicMock() + if wait_side_effect is not None: + proc.wait.side_effect = wait_side_effect + else: + proc.wait.return_value = None + proc.pid = 12345 + return proc + + +class TestRunHcatCmd: + def test_normal_flow_waits_and_notifies(self, main_module, tmp_path): + hash_file = str(tmp_path / "hashes.txt") + proc = _make_mock_proc() + + with ( + patch("hate_crack.main.subprocess.Popen", return_value=proc) as mock_popen, + patch.object(main_module, "lineCount", return_value=42), + patch("hate_crack.main._notify") as mock_notify, + ): + mock_notify.is_suppressed.return_value = False + mock_notify.get_settings.return_value = MagicMock(enabled=True) + mock_notify.start_tailer.return_value = None + main_module._run_hcat_cmd( + ["hashcat", "-m", "1000"], attack_name="Brute Force", hash_file=hash_file + ) + + mock_popen.assert_called_once() + proc.wait.assert_called_once() + proc.kill.assert_not_called() + mock_notify.notify_job_done.assert_called_once_with( + "Brute Force", 42, hash_file + ) + + def test_keyboard_interrupt_kills_process(self, main_module, tmp_path): + hash_file = str(tmp_path / "hashes.txt") + proc = _make_mock_proc(wait_side_effect=KeyboardInterrupt()) + + with ( + patch("hate_crack.main.subprocess.Popen", return_value=proc), + patch.object(main_module, "lineCount", return_value=0), + patch("hate_crack.main._notify") as mock_notify, + ): + mock_notify.is_suppressed.return_value = False + mock_notify.get_settings.return_value = MagicMock(enabled=False) + mock_notify.start_tailer.return_value = None + main_module._run_hcat_cmd( + ["hashcat"], attack_name="Brute Force", hash_file=hash_file + ) + + proc.kill.assert_called_once() + + def test_no_notify_when_attack_name_empty(self, main_module, tmp_path): + hash_file = str(tmp_path / "hashes.txt") + proc = _make_mock_proc() + + with ( + patch("hate_crack.main.subprocess.Popen", return_value=proc), + patch("hate_crack.main._notify") as mock_notify, + ): + mock_notify.is_suppressed.return_value = False + main_module._run_hcat_cmd(["hashcat"], attack_name="", hash_file=hash_file) + + mock_notify.notify_job_done.assert_not_called() + mock_notify.start_tailer.assert_not_called() + + def test_suppressed_skips_notifications(self, main_module, tmp_path): + hash_file = str(tmp_path / "hashes.txt") + proc = _make_mock_proc() + + with ( + patch("hate_crack.main.subprocess.Popen", return_value=proc), + patch("hate_crack.main._notify") as mock_notify, + ): + mock_notify.is_suppressed.return_value = True + mock_notify.get_settings.return_value = MagicMock(enabled=True) + main_module._run_hcat_cmd( + ["hashcat"], attack_name="Brute Force", hash_file=hash_file + ) + + mock_notify.start_tailer.assert_not_called() + mock_notify.notify_job_done.assert_not_called() + + def test_stdin_is_forwarded_to_popen(self, main_module, tmp_path): + stdin_stub = object() + proc = _make_mock_proc() + + with ( + patch("hate_crack.main.subprocess.Popen", return_value=proc) as mock_popen, + patch("hate_crack.main._notify") as mock_notify, + ): + mock_notify.is_suppressed.return_value = False + mock_notify.get_settings.return_value = MagicMock(enabled=False) + mock_notify.start_tailer.return_value = None + main_module._run_hcat_cmd(["hashcat"], stdin=stdin_stub) + + _, kwargs = mock_popen.call_args + assert kwargs.get("stdin") is stdin_stub + + def test_companion_procs_killed_on_interrupt(self, main_module, tmp_path): + hash_file = str(tmp_path / "hashes.txt") + proc = _make_mock_proc(wait_side_effect=KeyboardInterrupt()) + companion = _make_mock_proc() + + with ( + patch("hate_crack.main.subprocess.Popen", return_value=proc), + patch.object(main_module, "lineCount", return_value=0), + patch("hate_crack.main._notify") as mock_notify, + ): + mock_notify.is_suppressed.return_value = False + mock_notify.get_settings.return_value = MagicMock(enabled=False) + mock_notify.start_tailer.return_value = None + main_module._run_hcat_cmd( + ["hashcat"], + attack_name="Combinator3", + hash_file=hash_file, + companion_procs=[companion], + ) + + proc.kill.assert_called_once() + companion.kill.assert_called_once() + + def test_companion_procs_waited_on_normal_exit(self, main_module, tmp_path): + hash_file = str(tmp_path / "hashes.txt") + proc = _make_mock_proc() + companion = _make_mock_proc() + + with ( + patch("hate_crack.main.subprocess.Popen", return_value=proc), + patch("hate_crack.main._notify") as mock_notify, + ): + mock_notify.is_suppressed.return_value = False + mock_notify.get_settings.return_value = MagicMock(enabled=False) + mock_notify.start_tailer.return_value = None + main_module._run_hcat_cmd( + ["hashcat"], + attack_name="Combinator3", + hash_file=hash_file, + companion_procs=[companion], + ) + + companion.wait.assert_called_once() + companion.kill.assert_not_called() + + def test_reraise_interrupt_propagates(self, main_module, tmp_path): + hash_file = str(tmp_path / "hashes.txt") + proc = _make_mock_proc(wait_side_effect=KeyboardInterrupt()) + + with ( + patch("hate_crack.main.subprocess.Popen", return_value=proc), + patch.object(main_module, "lineCount", return_value=0), + patch("hate_crack.main._notify") as mock_notify, + ): + mock_notify.is_suppressed.return_value = False + mock_notify.get_settings.return_value = MagicMock(enabled=False) + mock_notify.start_tailer.return_value = None + with pytest.raises(KeyboardInterrupt): + main_module._run_hcat_cmd( + ["hashcat"], + attack_name="YOLO", + hash_file=hash_file, + reraise_interrupt=True, + ) + + def test_out_path_override(self, main_module, tmp_path): + hash_file = str(tmp_path / "hashes.txt") + alt_out = str(tmp_path / "hashes.lm.cracked") + proc = _make_mock_proc() + + with ( + patch("hate_crack.main.subprocess.Popen", return_value=proc), + patch.object(main_module, "lineCount", return_value=9) as mock_lc, + patch("hate_crack.main._notify") as mock_notify, + ): + mock_notify.is_suppressed.return_value = False + mock_notify.get_settings.return_value = MagicMock(enabled=True) + mock_notify.start_tailer.return_value = None + main_module._run_hcat_cmd( + ["hashcat"], + attack_name="LM Phase", + hash_file=hash_file, + out_path=alt_out, + ) + + mock_lc.assert_called_with(alt_out) + mock_notify.notify_job_done.assert_called_once_with( + "LM Phase", 9, hash_file + ) + + def test_tailer_is_stopped_in_finally(self, main_module, tmp_path): + hash_file = str(tmp_path / "hashes.txt") + proc = _make_mock_proc(wait_side_effect=KeyboardInterrupt()) + tailer = MagicMock() + + with ( + patch("hate_crack.main.subprocess.Popen", return_value=proc), + patch.object(main_module, "lineCount", return_value=0), + patch("hate_crack.main._notify") as mock_notify, + ): + mock_notify.is_suppressed.return_value = False + mock_notify.get_settings.return_value = MagicMock(enabled=True) + mock_notify.start_tailer.return_value = tailer + main_module._run_hcat_cmd( + ["hashcat"], attack_name="Brute Force", hash_file=hash_file + ) + + mock_notify.stop_tailer.assert_called_once_with(tailer) diff --git a/tests/test_ui_menu_options.py b/tests/test_ui_menu_options.py index 955121c..77f4294 100644 --- a/tests/test_ui_menu_options.py +++ b/tests/test_ui_menu_options.py @@ -32,6 +32,7 @@ MENU_OPTION_TEST_CASES = [ ("22", CLI_MODULE._attacks, "combipow_crack", "combipow"), ("80", CLI_MODULE._attacks, "wordlist_tools_submenu", "wordlist-tools"), ("81", CLI_MODULE._attacks, "rule_tools_submenu", "rule-tools"), + ("82", CLI_MODULE, "notifications_submenu", "notifications-submenu"), ("90", CLI_MODULE, "download_hashmob_rules", "hashmob-rules"), ("91", CLI_MODULE, "weakpass_wordlist_menu", "weakpass-menu"), ("92", CLI_MODULE, "download_hashmob_wordlists", "hashmob-wordlists"), @@ -77,3 +78,19 @@ def test_main_menu_option_94_hashview_visible_with_hashview_api_key(monkeypatch) options = CLI_MODULE.get_main_menu_options() assert "94" in options assert options["94"]() == sentinel + + +def test_main_menu_no_longer_exposes_options_83_84(): + """Options 83 and 84 moved into the Notifications submenu (option 82).""" + options = CLI_MODULE.get_main_menu_options() + assert "83" not in options + assert "84" not in options + assert "82" in options + + +def test_main_menu_items_include_notifications_entry(): + items = dict(CLI_MODULE.get_main_menu_items()) + assert "82" in items + assert "Notifications" in items["82"] + assert "83" not in items + assert "84" not in items diff --git a/tests/test_ui_test_pushover.py b/tests/test_ui_test_pushover.py new file mode 100644 index 0000000..f5b43b4 --- /dev/null +++ b/tests/test_ui_test_pushover.py @@ -0,0 +1,79 @@ +"""Unit tests for main.test_pushover_notification(). + +These tests patch ``hate_crack.main._notify`` directly rather than +``hate_crack.notify._send_pushover``. The latter is fragile because +``tests/test_random_rules_attack.py`` purges ``hate_crack.*`` from +``sys.modules`` and re-imports, which leaves ``main._notify`` pointing +at a different module object than the one a top-level ``patch`` would +touch. Patching the attribute on ``main`` itself is robust to that. + +We use ``patch.object(hc_main, "_notify")`` rather than +``patch("hate_crack.main._notify")`` so the patch target is the exact +module object whose function we invoke. A string target would re-resolve +``hate_crack.main`` through ``sys.modules``, which — again thanks to the +purge in ``test_random_rules_attack.py`` — can be a different object from +the ``hc_main`` reference bound at test-module import time. +""" + +from types import SimpleNamespace +from unittest.mock import patch + +from hate_crack import main as hc_main + + +def _settings( + *, enabled: bool = True, token: str = "tok", user: str = "usr" +) -> SimpleNamespace: + """Minimal stand-in for ``NotifySettings`` — we only read three fields.""" + return SimpleNamespace(enabled=enabled, pushover_token=token, pushover_user=user) + + +class TestTestPushoverNotification: + def test_success_prints_confirmation_and_sends(self, capsys): + with patch.object(hc_main, "_notify") as mock_notify: + mock_notify.get_settings.return_value = _settings(enabled=True) + mock_notify._send_pushover.return_value = True + hc_main.test_pushover_notification() + mock_notify._send_pushover.assert_called_once() + args = mock_notify._send_pushover.call_args.args + assert args[0] == "tok" + assert args[1] == "usr" + assert args[2] == "hate_crack: test notification" + assert "test notification from hate_crack" in args[3] + out = capsys.readouterr().out + assert "[+] Test Pushover notification sent" in out + + def test_failure_prints_failure_line(self, capsys): + with patch.object(hc_main, "_notify") as mock_notify: + mock_notify.get_settings.return_value = _settings(enabled=True) + mock_notify._send_pushover.return_value = False + hc_main.test_pushover_notification() + out = capsys.readouterr().out + assert "[!] Test Pushover notification failed" in out + + def test_missing_token_skips_send_and_warns(self, capsys): + with patch.object(hc_main, "_notify") as mock_notify: + mock_notify.get_settings.return_value = _settings(enabled=True, token="") + hc_main.test_pushover_notification() + mock_notify._send_pushover.assert_not_called() + out = capsys.readouterr().out + assert "[!] Pushover credentials missing" in out + assert "notify_pushover_token" in out + + def test_missing_user_skips_send_and_warns(self, capsys): + with patch.object(hc_main, "_notify") as mock_notify: + mock_notify.get_settings.return_value = _settings(enabled=True, user="") + hc_main.test_pushover_notification() + mock_notify._send_pushover.assert_not_called() + out = capsys.readouterr().out + assert "[!] Pushover credentials missing" in out + + def test_globally_off_still_sends_with_note(self, capsys): + with patch.object(hc_main, "_notify") as mock_notify: + mock_notify.get_settings.return_value = _settings(enabled=False) + mock_notify._send_pushover.return_value = True + hc_main.test_pushover_notification() + mock_notify._send_pushover.assert_called_once() + out = capsys.readouterr().out + assert "notifications are globally OFF" in out + assert "[+] Test Pushover notification sent" in out