gitea-mirror/hate_crack
1
0
Fork 0
mirror of https://github.com/trustedsec/hate_crack.git synced 2026-07-28 14:47:22 -07:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #109 from trustedsec/feat/pushover-notifications

Browse Source 
feat: add Pushover notifications for hashcat jobs
This commit is contained in:
Justin Bollinger
2026-04-22 20:42:45 -04:00
committed by GitHub
parent 28d442a63e 9eb584cda4
commit 4f1321cbe4
23 changed files with 2710 additions and 244 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.git-blame-ignore-revs
+9
View File
@@ -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
README.md
+17
View File
@@ -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.
config.json.example
+9 -1
View File
@@ -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
}
hate_crack.py
+1
View File
@@ -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,
hate_crack/attacks.py
+54 -16
View File
@@ -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)
hate_crack/config.json.example
+9 -1
View File
@@ -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
}
hate_crack/main.py
+378 -226
View File
@@ -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,
hate_crack/notify/__init__.py
+311
View File
@@ -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)
hate_crack/notify/_suppress.py
+38
View File
@@ -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
hate_crack/notify/pushover.py
+69
View File
@@ -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
hate_crack/notify/settings.py
+179
View File
@@ -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)
hate_crack/notify/tailer.py
+208
View File
@@ -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]
tests/conftest.py
+22
View File
@@ -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()
tests/test_notifications_submenu.py
+136
View File
@@ -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"]
tests/test_notify_integration.py
+242
View File
@@ -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)
tests/test_notify_per_crack_toggle.py
+135
View File
@@ -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
tests/test_notify_pushover.py
+88
View File
@@ -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
tests/test_notify_settings.py
+198
View File
@@ -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
tests/test_notify_suppression.py
+74
View File
@@ -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
tests/test_notify_tailer.py
+219
View File
@@ -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()
tests/test_run_hcat_cmd.py
+218
View File
@@ -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)
tests/test_ui_menu_options.py
+17
View File
@@ -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
tests/test_ui_test_pushover.py
+79
View File
@@ -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