diff --git a/hate_crack/main.py b/hate_crack/main.py index 19cba8c..e631957 100755 --- a/hate_crack/main.py +++ b/hate_crack/main.py @@ -1161,7 +1161,11 @@ def _write_field_sorted_unique(input_path, output_path, field_index, delimiter=" open(output_path, "w") as dst, ): sort_proc = subprocess.Popen( - ["sort", "-u"], stdin=subprocess.PIPE, stdout=dst, text=True + ["sort", "-u"], + stdin=subprocess.PIPE, + stdout=dst, + text=True, + env={**os.environ, "LC_ALL": "C"}, ) for line in src: line = line.rstrip("\n") @@ -1542,7 +1546,10 @@ def hcatFingerprint( if expander_stdout is None: raise RuntimeError("expander stdout pipe was not created") sort_proc = subprocess.Popen( - ["sort", "-u"], stdin=expander_stdout, stdout=dst + ["sort", "-u"], + stdin=expander_stdout, + stdout=dst, + env={**os.environ, "LC_ALL": "C"}, ) hcatProcess = sort_proc expander_stdout.close() @@ -2993,7 +3000,10 @@ def hcatLMtoNT(): stdout=subprocess.PIPE, ) hcatProcess = subprocess.Popen( - ["sort", "-u"], stdin=combine_proc.stdout, stdout=combined_out + ["sort", "-u"], + stdin=combine_proc.stdout, + stdout=combined_out, + env={**os.environ, "LC_ALL": "C"}, ) combine_proc.stdout.close() try: diff --git a/tests/test_sort_locale.py b/tests/test_sort_locale.py new file mode 100644 index 0000000..9a3d0d3 --- /dev/null +++ b/tests/test_sort_locale.py @@ -0,0 +1,163 @@ +"""Sort subprocess calls must force LC_ALL=C. + +macOS `sort` is locale-strict: when stdin contains bytes that aren't +valid in the current LC_COLLATE (commonly LC_COLLATE=en_US.UTF-8), +`sort` errors out with "sort: Illegal byte sequence". Cracked-password +streams routinely contain such bytes (hex-encoded fields, mixed +encodings, binary), so all three `sort -u` invocations in main.py must +override LC_ALL=C to fall back to byte-collation. +""" + +import importlib +import io +import subprocess + + +def _collect_sort_calls(monkeypatch, hc_main): + """Replace subprocess.Popen with a recorder. Returns a list that will + receive (args, kwargs) tuples for every Popen call so tests can pick + out the sort invocations and inspect their env.""" + calls = [] + + class RecordingPopen: + def __init__(self, args, **kwargs): + calls.append((args, kwargs)) + self.args = args + self.pid = 0 + self.stdout = None + self.stdin = None + cmd0 = args[0] if args else None + # Pipe behavior for the sort step so callers can read/write + # without blocking. + if cmd0 == "sort": + stdin = kwargs.get("stdin") + stdout = kwargs.get("stdout") + data = b"" + if stdin is not None and hasattr(stdin, "read"): + data = stdin.read() + if isinstance(data, str): + data = data.encode() + if stdout is not None and hasattr(stdout, "write"): + for ln in sorted(set(data.splitlines())): + stdout.write(ln + b"\n" if isinstance(ln, bytes) else ln + "\n") + if hasattr(stdout, "flush"): + stdout.flush() + # Sort also receives writes via stdin.write() in + # _write_field_sorted_unique. For that path we need a + # writable stdin handle. + if stdin is subprocess.PIPE or kwargs.get("text"): + self.stdin = io.StringIO() + elif isinstance(cmd0, str) and "expander" in cmd0: + stdin = kwargs.get("stdin") + data = stdin.read() if stdin is not None else b"" + self.stdout = io.BytesIO(data) + + def wait(self, timeout=None): + return 0 + + def kill(self): + pass + + monkeypatch.setattr(hc_main.subprocess, "Popen", RecordingPopen) + return calls + + +def _sort_calls(calls): + return [(args, kwargs) for args, kwargs in calls if args and args[0] == "sort"] + + +def test_write_field_sorted_unique_uses_C_locale(monkeypatch, tmp_path): + monkeypatch.setenv("HATE_CRACK_SKIP_INIT", "1") + import hate_crack.main as hc_main + + importlib.reload(hc_main) + + src = tmp_path / "input.txt" + src.write_text("aa:bb\ncc:dd\n") + dst = tmp_path / "out.txt" + + calls = _collect_sort_calls(monkeypatch, hc_main) + hc_main._write_field_sorted_unique(str(src), str(dst), 2) + + sort_calls = _sort_calls(calls) + assert sort_calls, "expected at least one sort -u invocation" + for _args, kwargs in sort_calls: + env = kwargs.get("env") + assert env is not None, "sort Popen must pass env to force locale" + assert env.get("LC_ALL") == "C", ( + f"sort env must set LC_ALL=C to handle non-UTF-8 bytes; got {env.get('LC_ALL')!r}" + ) + + +def test_hcatFingerprint_sort_uses_C_locale(monkeypatch, tmp_path): + monkeypatch.setenv("HATE_CRACK_SKIP_INIT", "1") + import hate_crack.main as hc_main + + importlib.reload(hc_main) + + hashfile = tmp_path / "hashes.txt" + out_path = tmp_path / "hashes.txt.out" + out_path.write_text("deadbeef:somepassword\n") + + monkeypatch.setattr(hc_main, "lineCount", lambda _p: 1) + monkeypatch.setattr(hc_main, "hcatHashCracked", 0) + monkeypatch.setattr(hc_main, "ensure_binary", lambda binary_path, **_k: binary_path) + monkeypatch.setattr(hc_main, "hcatHybrid", lambda *a, **kw: None) + monkeypatch.setattr(hc_main, "hcatHashFile", str(hashfile), raising=False) + + calls = _collect_sort_calls(monkeypatch, hc_main) + hc_main.hcatFingerprint( + "1000", str(hashfile), expander_len=7, run_hybrid_on_expanded=False + ) + + sort_calls = _sort_calls(calls) + assert sort_calls, "expected at least one sort -u invocation in fingerprint pipeline" + for _args, kwargs in sort_calls: + env = kwargs.get("env") + assert env is not None + assert env.get("LC_ALL") == "C" + + +def test_all_sort_popen_calls_in_main_set_LC_ALL_C(): + """Source-level guard: every subprocess.Popen call in main.py whose + first arg list begins with "sort" must also pass an env= kwarg that + sets LC_ALL=C. This catches the third site (hcatLMtoNT) without + needing to mock its substantial setup, and prevents future sort + invocations from regressing the locale fix.""" + import ast + import pathlib + + main_path = pathlib.Path(__file__).parent.parent / "hate_crack" / "main.py" + tree = ast.parse(main_path.read_text()) + + offenders = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + func = node.func + # Match subprocess.Popen(...) + if not ( + isinstance(func, ast.Attribute) + and func.attr == "Popen" + and isinstance(func.value, ast.Name) + and func.value.id == "subprocess" + ): + continue + if not node.args: + continue + first = node.args[0] + # First positional arg must be a list whose first element is "sort" + if not isinstance(first, ast.List) or not first.elts: + continue + head = first.elts[0] + if not (isinstance(head, ast.Constant) and head.value == "sort"): + continue + # Found a sort Popen. Require env= kwarg. + env_kw = next((kw for kw in node.keywords if kw.arg == "env"), None) + if env_kw is None: + offenders.append(node.lineno) + + assert not offenders, ( + f"subprocess.Popen([\"sort\", ...]) at line(s) {offenders} must pass " + "env={**os.environ, 'LC_ALL': 'C'} to handle non-UTF-8 bytes on macOS." + ) diff --git a/tests/test_utils.py b/tests/test_utils.py index df2f12f..fd688cf 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -191,7 +191,7 @@ def test_line_count_and_write_helpers(tmp_path, monkeypatch): ) class FakePopen: - def __init__(self, args, stdin=None, stdout=None, text=None): + def __init__(self, args, stdin=None, stdout=None, text=None, **_kwargs): self.stdin = FakeStdin(self) self._stdout = stdout self._data = None