diff --git a/plugins/disk-hygiene/.claude-plugin/plugin.json b/plugins/disk-hygiene/.claude-plugin/plugin.json index cced5dd66c..7e941283b5 100644 --- a/plugins/disk-hygiene/.claude-plugin/plugin.json +++ b/plugins/disk-hygiene/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "$schema": "https://json.schemastore.org/claude-code-plugin-manifest.json", "name": "disk-hygiene", - "version": "0.41.0", + "version": "0.41.1", "description": "Context-aware disk hygiene for arbitrary directory trees: inventories orphaned and temporary artifacts, classifies evidence into review tiers, and offers exact-path cleanup only after a fresh safety preview and explicit per-tier approval. The target is read-only by default; OS-managed paths, links and mount points, VCS-tracked content without the complete checkout evidence bundle, changed entries, and live-handle uncertainty fail closed.", "author": { "name": "Melodic Software", diff --git a/plugins/disk-hygiene/CHANGELOG.md b/plugins/disk-hygiene/CHANGELOG.md index d9f1e7a6bc..a3c96ef8bb 100644 --- a/plugins/disk-hygiene/CHANGELOG.md +++ b/plugins/disk-hygiene/CHANGELOG.md @@ -3,6 +3,12 @@ All notable changes to the `disk-hygiene` plugin are documented here. Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); this plugin uses semantic versioning. +## [0.41.1] - 2026-09-30 + +### Changed + +- **The engine-gate denial names what failed** ([#5519](https://github.com/melodic-software/claude-code-plugins/issues/5519)). A denied Bash command now says which token or stage the exact-engine classifier refused, states the flag-order rule (required flags first, in declared order, then optional flags in any order), and states which mention forms are gated and which read-only forms work, adding that a relative path or bare name which resolves to the installed engine from the current directory is still gated. A quoted word the gate reads as an engine call, such as a `gh --search` query naming an interpreter and the engine, is named as the gated word. The message is the only change: the guard matches and denies exactly as before. + ## [0.41.0] - 2026-09-30 ### Added diff --git a/plugins/disk-hygiene/lib/engine_grammar.py b/plugins/disk-hygiene/lib/engine_grammar.py index 36865269d3..08d7bfb54e 100644 --- a/plugins/disk-hygiene/lib/engine_grammar.py +++ b/plugins/disk-hygiene/lib/engine_grammar.py @@ -415,3 +415,108 @@ def value_ok(flag: Flag, value: str) -> bool: for flag in spec.optional if flag.requires is not None and flag.name in seen ) and all(len(seen.intersection(group)) == 1 for group in spec.one_of) + + +_TOKEN_CAP = 80 + + +def clip_token(value: str) -> str: + """A user-supplied token quoted for a message, capped so a paste stays short.""" + clipped = value if len(value) <= _TOKEN_CAP else value[: _TOKEN_CAP - 3] + "..." + return repr(clipped) + + +def required_order(spec: Subcommand) -> str: + """The required head of ``spec`` spelled in the one order the guard admits.""" + return ", ".join(flag.name for flag in spec.required) + + +def _order_rule(spec: Subcommand) -> str: + if not spec.required: + return "every flag is optional and may come in any order" + return ( + f"required flags first, in order: {required_order(spec)}; " + "then optional flags in any order" + ) + + +def explain_mismatch( + name: str, + words: list[str], + external_checks: dict[str, object] | None = None, +) -> str | None: + """Name the first word ``match_invocation`` refuses and the rule it broke. + + Runs on the deny path only and walks the same grammar in the same order, so + it returns ``None`` exactly when ``match_invocation`` returns ``True``. + """ + spec = subcommand(name) + if spec is None: + return f"{clip_token(name)} is not an engine subcommand." + checks = external_checks or {} + + def value_problem(flag: Flag, value: str) -> str | None: + subject = f"{flag.name} value {clip_token(value)}" + if not is_argument(value): + return f"{subject} must be a literal, not empty or starting with '-'." + if flag.choices is not None and value not in flag.choices: + return f"{subject} must be one of: {', '.join(sorted(flag.choices))}." + if flag.pattern is not None and flag.pattern.fullmatch(value) is None: + return f"{subject} must match {flag.pattern.pattern}." + if flag.external_check is not None: + check = checks.get(flag.external_check) + if not (callable(check) and check(value)): + return f"{subject} is not the {flag.external_check}." + return None + + def read_value(flag: Flag, index: int) -> tuple[int, str | None]: + if not flag.takes_value: + return index, None + if index >= len(words): + return index, f"{flag.name} needs a value." + return index + 1, value_problem(flag, words[index]) + + index = 0 + for flag in spec.required: + if index >= len(words): + return f"required flag {flag.name} is missing; {_order_rule(spec)}." + if words[index] != flag.name: + return ( + f"{clip_token(words[index])} is where required flag {flag.name} " + f"belongs; {_order_rule(spec)}." + ) + index, problem = read_value(flag, index + 1) + if problem: + return problem + + seen: set[str] = set() + while index < len(words): + word = words[index] + flag = spec.flag(word) + if flag is None: + return f"{clip_token(word)} is not a {name} flag." + if flag.required: + return ( + f"required flag {flag.name} is already given in the required head " + "and cannot repeat." + ) + if flag.name in seen and not flag.repeatable: + return f"{flag.name} is not repeatable but is given twice." + seen.add(flag.name) + index, problem = read_value(flag, index + 1) + if problem: + return problem + + for flag in spec.optional: + if ( + flag.requires is not None + and flag.name in seen + and flag.requires not in seen + ): + return f"{flag.name} requires {flag.requires}." + for group in spec.one_of: + given = sorted(seen.intersection(group)) + if len(given) != 1: + count = "none was" if not given else f"{len(given)} were" + return f"give exactly one of {', '.join(group)}; {count} given." + return None diff --git a/plugins/disk-hygiene/skills/clean/scripts/destructive_guard.py b/plugins/disk-hygiene/skills/clean/scripts/destructive_guard.py index 5b9ed3ce41..87349ea350 100755 --- a/plugins/disk-hygiene/skills/clean/scripts/destructive_guard.py +++ b/plugins/disk-hygiene/skills/clean/scripts/destructive_guard.py @@ -698,6 +698,29 @@ def _carries_marker(word: str) -> bool: return name == _ENGINE_MARKER +def _is_interpreter(word: str) -> bool: + base = Path(word.casefold()).name + return base.startswith("python") or base in {"py", "py.exe"} + + +def _reads_as_engine_payload(word: str) -> bool: + """Whether a word that is not the engine path still reads as an engine call. + + A quoted compound payload (sh -c / pwsh -Command) whose first token is the + engine or an interpreter, or any word holding both the engine filename and + "python". The gate and its denial reason share this test, so the reason + names the word the gate acted on. + """ + folded = word.casefold() + if _carries_marker(word) or _ENGINE_MARKER not in folded: + return False + if " " in word: + first_token = folded.split()[0] + if _carries_marker(first_token) or _is_interpreter(first_token): + return True + return "python" in folded + + # PowerShell also closes a quote opened with ' by any of these and a " by the # double forms; a scanner that knows only the ASCII pair would see a different # string boundary than PowerShell does. @@ -875,10 +898,6 @@ def _engine_gate_relevant(command: str, tool_name: str = "Bash") -> bool: the rest of the session once that belt has registered). """ - def _is_interpreter(word: str) -> bool: - base = Path(word.casefold()).name - return base.startswith("python") or base in {"py", "py.exe"} - bundled = _engine_script_path() def _samefile(word: str) -> bool: @@ -1022,7 +1041,6 @@ def _provably_other_file(token: str, word: str) -> bool: if Path(word.casefold()).name in _WRAPPERS ] for index, word in enumerate(words): - folded = word.casefold() if _carries_marker(word): if _samefile(word) or _within_plugin_cache_family(word): # The word is one of THIS PLUGIN'S engines — the bundled one, @@ -1056,13 +1074,7 @@ def _provably_other_file(token: str, word: str) -> bool: # (`git diff -- hygiene.py`) still defer. return True continue - if _ENGINE_MARKER in folded and " " in word: - first_token = folded.split()[0] - if _carries_marker(first_token) or _is_interpreter(first_token): - # A quoted compound payload (sh -c / pwsh -Command) whose first - # token is the engine or an interpreter is an invocation. - return True - if _ENGINE_MARKER in folded and "python" in folded: + if _reads_as_engine_payload(word): return True return False @@ -2148,7 +2160,145 @@ def _bash_allowlist_disclosure(authority: str | None) -> str: ) -def _bash_denial_guidance(authority: str | None, mode: str | None = None) -> str: +_OPERATOR_LABELS = { + char: label + for chars, label in ( + ("|", "a pipe"), + ("<>", "a redirect"), + (";", "a ';'"), + ("&", "an '&'"), + ("$`(){}", "a substitution or expansion"), + ("*?[]~", "a glob or tilde"), + ("\r\n\t", "a newline or tab"), + ("!#", "a '!' or '#'"), + ) + for char in chars +} + + +def _unparsable_reason(command: str) -> str: + """Name the first thing in ``command`` that ``_literal_shell_words`` rejects.""" + for char in command: + if char in _OPERATOR_LABELS: + culprit = f"{_OPERATOR_LABELS[char]} ({char!r})" + break + else: + culprit = ( + "a backslash" + if "\\" in command + else "a quote that does not wrap a whole word, or a non-space whitespace character" + ) + return f"The command contains {culprit}, so it is not one plain literal invocation." + + +def _resolves_to_engine(word: str) -> bool: + key = _script_path_key(word) + return key is not None and key == _script_path_key(str(_engine_script_path())) + + +def _engine_mismatch_reason(command: str, authority: str | None) -> str: + """One sentence naming what ``classify_exact_engine_command`` refuses. + + Deny path only; the caller has already denied and this decides nothing. It + walks the classifier's stages in order, except that when the command is not + the hook's Python, an engine operand or a word that reads as an engine + payload is named first: that word is what the gate acts on, whatever the + command's length. + """ + tokens = _literal_shell_words(command) + if tokens is None: + return _unparsable_reason(command) + python_ok = _is_current_python(tokens[0]) + operand = ( + None + if python_ok + else next((word for word in tokens[1:] if _resolves_to_engine(word)), None) + ) + payload = ( + None + if python_ok or operand is not None + else next((word for word in tokens if _reads_as_engine_payload(word)), None) + ) + if payload is not None: + return ( + f"{engine_grammar.clip_token(payload)} holds the engine filename with " + "an interpreter or as its first word, so the gate reads it as an " + "engine call." + ) + if operand is not None: + named = ( + "is the engine path" + if os.path.isabs(operand) + else "resolves to the engine from the current directory" + ) + return ( + f"{engine_grammar.clip_token(operand)} {named}, and only a call " + f'through "{_display_python()}" may name it; ' + f"{engine_grammar.clip_token(tokens[0])} is not that interpreter." + ) + if len(tokens) < 3: + return ( + f"The command has {len(tokens)} word(s); an engine call is " + " ." + ) + if not python_ok: + return ( + f"{engine_grammar.clip_token(tokens[0])} is not this hook's Python; " + f'the interpreter must be "{_display_python()}".' + ) + if not _resolves_to_engine(tokens[1]): + return ( + f"{engine_grammar.clip_token(tokens[1])} is not the bundled engine " + f'"{_display_path(_engine_script_path())}".' + ) + subcommand = tokens[2] + if subcommand not in _ALLOWED_ENGINE_SUBCOMMANDS: + return ( + f"{engine_grammar.clip_token(subcommand)} is not an engine " + f"subcommand; use one of {', '.join(_ALLOWED_ENGINE_SUBCOMMANDS)}." + ) + words = tokens[3:] + if engine_grammar.DATA_ROOT_FLAG not in words: + return ( + f"{engine_grammar.DATA_ROOT_FLAG} is missing; every engine call " + f"passes {engine_grammar.DATA_ROOT_FLAG} with the authorized root." + ) + external_checks = { + engine_grammar.AUTHORIZED_DATA_ROOT: ( + lambda value: _is_authorized_data_root(value, authority) + ), + } + return ( + engine_grammar.explain_mismatch(subcommand, words, external_checks) + or "The arguments do not match the engine grammar." + ) + + +def _engine_flag_order_rule() -> str: + heads = "; ".join( + f"{spec.name}: {engine_grammar.required_order(spec)}" + for spec in engine_grammar.SUBCOMMANDS + if spec.required + ) + return ( + "Flag order: required flags come first, in declared order " + f"({heads}), then optional flags in any order." + ) + + +_ENGINE_GATE_SCOPE = ( + "Any command that contains the engine filename together with a pipe, " + "redirect, ;, substitution, or an absolute engine-path operand is gated. " + "The read-only forms that work name the engine by a relative path or bare " + "name in a plain git show, git grep, grep or rg with no pipe, redirect or ;. " + "A relative path or bare name that resolves to the installed engine from " + "the current directory is still gated." +) + + +def _bash_denial_guidance( + authority: str | None, mode: str | None = None, command: str | None = None +) -> str: """Explain a Bash deny in the words of the surface that issued it. ``engine-gate`` (the plugin-level always-on hook) gates this engine @@ -2158,16 +2308,25 @@ def _bash_denial_guidance(authority: str | None, mode: str | None = None) -> str that skill is invoked, and names how it clears. Both bodies disclose the same classifier allow-list so the denial cannot teach a grammar the classifier does not implement. Unrecognized ``mode`` values fall back to - ``belt``, matching ``resolve_mode``. + ``belt``, matching ``resolve_mode``. ``command`` is read only by the + ``engine-gate`` body, to name what failed in the denied command. """ resolved = resolve_mode() if mode is None else mode grammar = _bash_allowlist_disclosure(authority) if resolved == _MODE_ENGINE_GATE: + reason = ( + f"{_engine_mismatch_reason(command, authority)} " + if command is not None + else "" + ) return ( "Disk-hygiene engine gate: this specific engine invocation is " - "gated. The rest of the Bash lane is unaffected, and " + "gated. " + reason + "The rest of the Bash lane is unaffected, and " "/disk-hygiene:clean need not have been invoked for this to fire. " - "Allowed shapes for this invocation are " + + _engine_flag_order_rule() + + " " + + _ENGINE_GATE_SCOPE + + " Allowed shapes for this invocation are " + grammar + " Supporting inspection of this invocation may use that small " "Bash allowlist or non-Bash read-only tools; any other shape of " @@ -2849,7 +3008,7 @@ def _decide(command: str, tool_name: str, start: float) -> int: start, "deny", "not-exact-engine-command", - _bash_denial_guidance(authority), + _bash_denial_guidance(authority, command=command), ) diff --git a/plugins/disk-hygiene/skills/clean/scripts/test_hygiene.py b/plugins/disk-hygiene/skills/clean/scripts/test_hygiene.py index 225f73ed4e..e58d616247 100755 --- a/plugins/disk-hygiene/skills/clean/scripts/test_hygiene.py +++ b/plugins/disk-hygiene/skills/clean/scripts/test_hygiene.py @@ -10464,6 +10464,174 @@ def test_bash_denial_modes_frame_opposite_scopes(self) -> None: self.assertIn("this specific engine invocation", gated_reason) self.assertNotIn("Bash is restricted", gated_reason) + def _engine_words(self, tail: str) -> str: + script = (SCRIPT_DIR / "hygiene.py").resolve().as_posix() + root = self._data_root.resolve().as_posix() + return f'"{self.python_command()}" "{script}" {tail} --data-root "{root}"' + + def _gated_reason(self, command: str) -> str: + result = self.run_guard_engine_gate(command) + assert result is not None + output = result["hookSpecificOutput"] + self.assertEqual("deny", output["permissionDecision"]) + return cast(str, output["permissionDecisionReason"]) + + def test_engine_gate_denial_names_the_flag_order_violation(self) -> None: + command = self._engine_words("scan --target t --root-children --output s") + reason = self._gated_reason(command) + self.assertIn("'--root-children'", reason) + self.assertIn("required flags first", reason) + self.assertIn("--target, --output", reason) + self.assertIn(guard._ENGINE_GATE_SCOPE, reason) + self.assertIn("read-only forms that work", reason) + # belt-mode text is unchanged by the reason + belt = guard._bash_denial_guidance("/data/root", mode=guard._MODE_BELT) + self.assertEqual( + belt, + guard._bash_denial_guidance( + "/data/root", mode=guard._MODE_BELT, command=command + ), + ) + self.assertNotIn("required flags first", belt) + + def test_engine_mismatch_reason_names_each_early_stage(self) -> None: + script = guard._display_path(guard._engine_script_path()) + python = self.python_command() + root = "/data/root" + tail = f'scan --target t --output s --data-root "{root}"' + cases = { + "wrong interpreter": ( + f'"/usr/bin/other" "{script}" {tail}', + "/usr/bin/other", + ), + "wrong script": (f'"{python}" "/tmp/other.py" {tail}', "/tmp/other.py"), + "unknown subcommand": ( + f'"{python}" "{script}" bogus --data-root "{root}"', + "bogus", + ), + "missing data-root": ( + f'"{python}" "{script}" scan --target t --output s', + "--data-root is missing", + ), + "unauthorized data-root": ( + f'"{python}" "{script}" scan --target t --output s ' + "--data-root /elsewhere", + "/elsewhere", + ), + "pipe after invocation": ( + f'"{python}" "{script}" {tail} | tail', + "pipe", + ), + } + for label, (command, expected) in cases.items(): + with self.subTest(label): + self.assertIn(expected, guard._engine_mismatch_reason(command, root)) + + def test_engine_mismatch_reason_names_the_engine_operand_of_a_mention( + self, + ) -> None: + engine = guard._display_path(guard._engine_script_path()) + for command in (f'grep foo "{engine}"', f'cat "{engine}"'): + with self.subTest(command): + reason = guard._engine_mismatch_reason(command, "/data/root") + self.assertIn("is the engine path", reason) + self.assertIn(command.split()[0], reason) + self.assertNotIn("not this hook's Python", reason) + plugin_root = guard._engine_script_path().parents[3] + relative = guard._engine_script_path().relative_to(plugin_root).as_posix() + with chdir_context(plugin_root): + reason = guard._engine_mismatch_reason(f"grep foo {relative}", "/data/root") + self.assertIn("resolves to the engine from the current directory", reason) + self.assertIn(relative, reason) + + def test_engine_mismatch_reason_names_only_the_operator_present(self) -> None: + script = guard._display_path(guard._engine_script_path()) + python = self.python_command() + head = f'"{python}" "{script}" scan --target t --output s --data-root /d' + cases = { + "pipe": (f"{head} | tail", "a pipe"), + "redirect": (f"{head} > out", "a redirect"), + "semicolon": (f"{head}; echo", "a ';'"), + "substitution": (f"{head} $(id)", "a substitution or expansion"), + "backslash": (f"{head} a\\b", "a backslash"), + "quote": (f'{head} a"b"', "a quote that does not wrap a whole word"), + } + for label, (command, expected) in cases.items(): + with self.subTest(label): + reason = guard._engine_mismatch_reason(command, "/d") + self.assertIn(expected, reason) + for other, (_, phrase) in cases.items(): + if other != label: + self.assertNotIn(phrase, reason) + + def test_operator_labels_cover_the_characters_the_literal_parser_rejects( + self, + ) -> None: + self.assertEqual( + set(guard._OPERATOR_LABELS), set(guard._SHELL_EXPANSION_OR_OPERATOR_CHARS) + ) + + def test_engine_gate_defers_the_read_only_forms_the_denial_advertises(self) -> None: + engine = guard._display_path(guard._engine_script_path()) + relative = "plugins/disk-hygiene/skills/clean/scripts/hygiene.py" + with tempfile.TemporaryDirectory() as tmp, chdir_context(tmp): + for command in ( + f"git show origin/main:{relative}", + "git grep foo -- hygiene.py", + f"grep foo {relative}", + "rg foo hygiene.py", + ): + self.assertFalse(guard._engine_gate_relevant(command, "Bash"), command) + for command in ( + f'grep foo "{engine}"', + f'cat "{engine}"', + f'git show "{engine}"', + f"grep foo {relative} | tail", + ): + self.assertTrue(guard._engine_gate_relevant(command, "Bash"), command) + + def test_engine_gate_gates_a_relative_word_that_resolves_to_the_engine( + self, + ) -> None: + engine = guard._engine_script_path() + plugin_root = engine.parents[3] + relative = engine.relative_to(plugin_root).as_posix() + rev_form = f"git show origin/main:{relative}" + cases = ( + (engine.parent, "rg foo hygiene.py"), + (engine.parent, "git grep foo -- hygiene.py"), + (plugin_root, f"grep foo {relative}"), + ) + for cwd, command in cases: + with self.subTest(cwd=cwd.name, command=command), chdir_context(cwd): + self.assertTrue(guard._engine_gate_relevant(command, "Bash")) + self.assertFalse(guard._engine_gate_relevant(rev_form, "Bash")) + self.assertIn( + "resolves to the installed engine from the current directory " + "is still gated", + self._gated_reason(command), + ) + + def test_engine_gate_denial_names_the_payload_word_it_gated(self) -> None: + cases = ( + 'gh issue list --search "python3 hygiene.py"', + 'gh issue list --search "hygiene.py scan"', + 'echo "run python3 hygiene.py"', + '"python3 hygiene.py scan"', + ) + with tempfile.TemporaryDirectory() as tmp, chdir_context(tmp): + for command in cases: + with self.subTest(command): + self.assertTrue(guard._engine_gate_relevant(command, "Bash")) + reason = self._gated_reason(command) + word = next(w for w in command.split('"') if "hygiene.py" in w) + self.assertIn(f"{word!r} holds the engine filename", reason) + self.assertNotIn("not this hook's Python", reason) + self.assertNotIn("word(s)", reason) + self.assertFalse( + guard._engine_gate_relevant("gh issue list --search hygiene.py", "Bash") + ) + def test_guard_allows_literal_readonly_supporting_bash_commands(self) -> None: """Belt inspection allowlist (#2591): read-only shapes pass; mutations stay denied. @@ -15416,6 +15584,72 @@ def test_parser_declares_exactly_the_grammar_flags(self) -> None: if flag.choices is not None: self.assertEqual(sorted(flag.choices), action.choices) + def _external_checks(self) -> dict[str, object]: + return {self.grammar.AUTHORIZED_DATA_ROOT: lambda v: v == self.AUTHORITY} + + def assert_refused(self, name: str, words: list[str], *named: str) -> None: + checks = self._external_checks() + reason = self.grammar.explain_mismatch(name, words, checks) + self.assertIsNotNone(reason, words) + self.assertFalse(self.grammar.match_invocation(name, words, checks), words) + for word in named: + self.assertIn(word, cast(str, reason)) + + def test_explainer_and_matcher_agree_on_every_declared_shape(self) -> None: + checks = self._external_checks() + for spec in self.grammar.SUBCOMMANDS: + for optionals in (False, True): + words = self.words(spec, optionals=optionals) + with self.subTest(subcommand=spec.name, optionals=optionals): + self.assertIsNone( + self.grammar.explain_mismatch(spec.name, words, checks) + ) + self.assertTrue( + self.grammar.match_invocation(spec.name, words, checks) + ) + + def test_explainer_names_the_offending_word(self) -> None: + scan = self.grammar.subcommand("scan") + assert scan is not None + base = self.words(scan, optionals=False) + head = self.head(scan) + # wrong order: an optional flag ahead of the required head + self.assert_refused( + "scan", + ["--root-children", *base], + "--root-children", + "required flags first", + "--target, --output", + ) + # unknown flag + self.assert_refused("scan", [*base, "--bogus"], "--bogus") + # duplicate of a non-repeatable flag + self.assert_refused( + "scan", [*base, "--policy", "p", "--policy", "q"], "--policy" + ) + # bad value + self.assert_refused("scan", [*base, "--max-depth", "0"], "--max-depth", "'0'") + self.assert_refused( + "scan", [*head, "--data-root", "/other"], "--data-root", "'/other'" + ) + # missing requires + self.assert_refused( + "scan", [*base, "--root-child", "x"], "--root-child", "--root-children" + ) + # missing required flag + self.assert_refused("scan", ["--target", "t"], "--output") + # missing one_of member + for spec in self.grammar.SUBCOMMANDS: + for group in spec.one_of: + required = [ + word for chunk in self.required_chunks(spec) for word in chunk + ] + self.assert_refused( + spec.name, + [*required, *self.data_root_chunk(spec)], + *group, + ) + def test_engine_parses_and_guard_admits_every_declared_shape(self) -> None: for spec in self.grammar.SUBCOMMANDS: for optionals in (False, True):