feat(pcfg): add hcatPrinceLing attack (mode B — cached wordlist via prince_ling, delegates to hcatPrince)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

hate_crack/main.py

@@ -2654,6 +2654,70 @@ def hcatPCFG(hcatHashType, hcatHashFile):
         pcfg_proc.stdout.close()
 
 
+def hcatPrinceLing(hcatHashType, hcatHashFile):
+    """Mode B: prince_ling generates a wordlist (with cache+staleness check),
+    then we delegate to the existing hcatPrince attack with hcatPrinceBaseList
+    temporarily rebound to the cached wordlist.
+    """
+    global hcatPrinceBaseList
+    pcfg_root = os.path.join(hate_path, "pcfg_cracker")
+    prince_ling_script = os.path.join(pcfg_root, "prince_ling.py")
+    ruleset_dir = os.path.join(pcfg_root, "Rules", pcfgRuleset)
+    if not os.path.isfile(prince_ling_script):
+        print(f"prince_ling.py not found at {prince_ling_script}")
+        return
+    if not os.path.isdir(ruleset_dir):
+        print(f"PCFG ruleset not found: {ruleset_dir}")
+        return
+
+    cache_dir = hcatOptimizedWordlists if isinstance(hcatOptimizedWordlists, str) \
+        else str(hcatOptimizedWordlists)
+    os.makedirs(cache_dir, exist_ok=True)
+    cache_path = os.path.join(cache_dir, f"pcfg_prince_ling_{pcfgRuleset}.txt")
+    tmp_path = cache_path + ".tmp"
+
+    # Staleness check: regenerate iff ruleset dir mtime > cache mtime (strict)
+    needs_regen = True
+    if os.path.isfile(cache_path):
+        ruleset_mtime = os.path.getmtime(ruleset_dir)
+        cache_mtime = os.path.getmtime(cache_path)
+        if ruleset_mtime <= cache_mtime:
+            needs_regen = False
+
+    if needs_regen:
+        print(f"[*] Generating prince_ling wordlist -> {cache_path}")
+        cmd = [
+            "python3",
+            prince_ling_script,
+            "--rule",
+            pcfgRuleset,
+            "--output",
+            tmp_path,
+            "--size",
+            str(pcfgPrinceLingMaxCandidates),
+        ]
+        try:
+            subprocess.run(cmd, check=True)
+            os.replace(tmp_path, cache_path)
+        except (subprocess.CalledProcessError, KeyboardInterrupt) as e:
+            # Clean up partial tmp file
+            if os.path.isfile(tmp_path):
+                try:
+                    os.remove(tmp_path)
+                except OSError:
+                    pass
+            print(f"prince_ling generation failed: {e}")
+            return
+
+    # Delegate to existing PRINCE attack with rebound base list
+    original_base = hcatPrinceBaseList
+    hcatPrinceBaseList = [cache_path]
+    try:
+        hcatPrince(hcatHashType, hcatHashFile)
+    finally:
+        hcatPrinceBaseList = original_base
+
+
 def hcatPermute(hcatHashType, hcatHashFile, wordlist):
     global hcatProcess, hcatPermuteCount
     permute_path = os.path.join(hate_path, "hashcat-utils", "bin", "permute.bin")

tests/test_main_pcfg.py

@@ -1,4 +1,5 @@
 """Tests for PCFG attack subprocess construction in hate_crack.main."""
+import os
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -59,3 +60,111 @@ class TestHcatPCFG:
         assert kwargs["stdin"] is not None
         assert kwargs["companion_procs"] is not None
         assert len(kwargs["companion_procs"]) == 1
+
+
+class TestHcatPrinceLing:
+    def _setup_pcfg_dirs(self, tmp_path, main_module, monkeypatch):
+        """Lay out fake pcfg_cracker/Rules/<ruleset>/ and optimized_wordlists/."""
+        pcfg_root = tmp_path / "pcfg_cracker"
+        rules_dir = pcfg_root / "Rules" / "DEFAULT"
+        rules_dir.mkdir(parents=True)
+        (rules_dir / "config.txt").write_text("dummy")
+        # prince_ling script must "exist" for the function to proceed
+        (pcfg_root / "prince_ling.py").write_text("# stub")
+        opt_dir = tmp_path / "optimized_wordlists"
+        opt_dir.mkdir()
+
+        monkeypatch.setattr(main_module, "hate_path", str(tmp_path))
+        monkeypatch.setattr(main_module, "hcatOptimizedWordlists", str(opt_dir))
+        return rules_dir, opt_dir
+
+    def test_regenerates_when_cache_stale(self, main_module, tmp_path, monkeypatch):
+        rules_dir, opt_dir = self._setup_pcfg_dirs(tmp_path, main_module, monkeypatch)
+        cache = opt_dir / "pcfg_prince_ling_DEFAULT.txt"
+        # Cache exists but is older than ruleset
+        cache.write_text("stale")
+        old = (rules_dir.stat().st_mtime - 100)
+        os.utime(cache, (old, old))
+
+        run_calls = []
+
+        def fake_run(cmd, **kwargs):
+            run_calls.append(cmd)
+            # Simulate prince_ling writing the .tmp file
+            for i, part in enumerate(cmd):
+                if part == "--output":
+                    Path(cmd[i + 1]).write_text("regenerated")
+            class R:
+                returncode = 0
+            return R()
+
+        with patch("hate_crack.main.subprocess.run", side_effect=fake_run), \
+             patch("hate_crack.main.hcatPrince") as mock_prince:
+            main_module.hcatPrinceLing("0", str(tmp_path / "hashes.txt"))
+
+        # prince_ling subprocess.run was invoked
+        assert len(run_calls) == 1
+        cmd = run_calls[0]
+        assert any("prince_ling.py" in p for p in cmd)
+        assert "--rule" in cmd
+        assert cmd[cmd.index("--rule") + 1] == "DEFAULT"
+        # Uses --size, NOT --limit
+        assert "--size" in cmd
+        assert "--limit" not in cmd
+        # hcatPrince delegated
+        assert mock_prince.called
+
+    def test_skips_regen_when_cache_fresh(self, main_module, tmp_path, monkeypatch):
+        rules_dir, opt_dir = self._setup_pcfg_dirs(tmp_path, main_module, monkeypatch)
+        cache = opt_dir / "pcfg_prince_ling_DEFAULT.txt"
+        cache.write_text("fresh")
+        # Cache is newer than ruleset
+        future = rules_dir.stat().st_mtime + 1000
+        os.utime(cache, (future, future))
+
+        with patch("hate_crack.main.subprocess.run") as mock_run, \
+             patch("hate_crack.main.hcatPrince"):
+            main_module.hcatPrinceLing("0", str(tmp_path / "hashes.txt"))
+
+        # subprocess.run was NOT called for prince_ling
+        assert not mock_run.called
+
+    def test_atomic_cache_write_cleans_tmp_on_failure(self, main_module, tmp_path, monkeypatch):
+        import subprocess as real_subprocess
+        rules_dir, opt_dir = self._setup_pcfg_dirs(tmp_path, main_module, monkeypatch)
+
+        def boom(cmd, **kwargs):
+            # Touch the .tmp file then fail (simulates partial write + crash)
+            for i, part in enumerate(cmd):
+                if part == "--output":
+                    Path(cmd[i + 1]).write_text("partial")
+            raise real_subprocess.CalledProcessError(1, cmd)
+
+        with patch("hate_crack.main.subprocess.run", side_effect=boom), \
+             patch("hate_crack.main.hcatPrince"):
+            main_module.hcatPrinceLing("0", str(tmp_path / "hashes.txt"))
+
+        # No real cache file created; tmp file cleaned up
+        assert not (opt_dir / "pcfg_prince_ling_DEFAULT.txt").exists()
+        assert not (opt_dir / "pcfg_prince_ling_DEFAULT.txt.tmp").exists()
+
+    def test_restores_hcatPrinceBaseList_on_exception(self, main_module, tmp_path, monkeypatch):
+        rules_dir, opt_dir = self._setup_pcfg_dirs(tmp_path, main_module, monkeypatch)
+        cache = opt_dir / "pcfg_prince_ling_DEFAULT.txt"
+        cache.write_text("fresh")
+        future = rules_dir.stat().st_mtime + 1000
+        os.utime(cache, (future, future))
+
+        original = ["original_base.txt"]
+        monkeypatch.setattr(main_module, "hcatPrinceBaseList", original)
+
+        def boom(*a, **kw):
+            raise RuntimeError("hcatPrince exploded")
+
+        with patch("hate_crack.main.hcatPrince", side_effect=boom):
+            try:
+                main_module.hcatPrinceLing("0", str(tmp_path / "hashes.txt"))
+            except RuntimeError:
+                pass
+
+        assert main_module.hcatPrinceBaseList == original