catenary-mcp 1.6.1

#!/usr/bin/env python3
"""Unit tests for the constrained_bash hook."""

import sys
import os
import unittest

# Add scripts/ to the import path so constrained_bash can be imported directly.
sys.path.insert(0, os.path.dirname(__file__))

from constrained_bash import check, deny_response


class TestAllowed(unittest.TestCase):
    def test_make(self):
        self.assertIsNone(check("make check"))

    def test_git(self):
        self.assertIsNone(check("git status"))
        self.assertIsNone(check("git log --oneline"))
        self.assertIsNone(check("git commit -m 'fix bug'"))

    def test_gh(self):
        self.assertIsNone(check("gh pr list"))
        self.assertIsNone(check("gh issue view 123"))

    def test_sleep(self):
        self.assertIsNone(check("sleep 5"))

    def test_cp_mv(self):
        self.assertIsNone(check("cp foo bar"))
        self.assertIsNone(check("mv foo bar"))

    def test_env_prefix_allowed(self):
        self.assertIsNone(check("DEBUG=1 make check"))
        self.assertIsNone(check("RUST_LOG=debug make test"))


class TestDenied(unittest.TestCase):
    def test_cat(self):
        self.assertIsNotNone(check("cat file.txt"))

    def test_grep(self):
        self.assertIsNotNone(check("grep foo bar.rs"))

    def test_ls(self):
        self.assertIsNotNone(check("ls -la"))

    def test_find(self):
        self.assertIsNotNone(check("find . -name '*.rs'"))

    def test_cargo(self):
        self.assertIsNotNone(check("cargo build"))
        self.assertIsNotNone(check("cargo test"))
        self.assertIsNotNone(check("cargo build 2>&1"))

    def test_full_path(self):
        self.assertIsNotNone(check("/usr/bin/grep foo bar"))
        self.assertIsNotNone(check("/bin/cat file.txt"))

    def test_env_prefix_denied(self):
        self.assertIsNotNone(check("DEBUG=1 cargo test"))


class TestGitDeniedSubcommands(unittest.TestCase):
    def test_git_grep(self):
        self.assertIsNotNone(check("git grep foo"))

    def test_git_ls_files(self):
        self.assertIsNotNone(check("git ls-files"))

    def test_git_ls_tree(self):
        self.assertIsNotNone(check("git ls-tree HEAD"))

    def test_git_log_allowed(self):
        self.assertIsNone(check("git log --oneline"))

    def test_git_diff_allowed(self):
        self.assertIsNone(check("git diff HEAD"))


class TestPipeline(unittest.TestCase):
    def test_grep_mid_pipeline(self):
        self.assertIsNone(check("gh pr list | grep foo"))

    def test_head_mid_pipeline(self):
        self.assertIsNone(check("gh issue list | head -20"))

    def test_tail_mid_pipeline(self):
        self.assertIsNone(check("git log --oneline | tail -5"))

    def test_jq_mid_pipeline(self):
        self.assertIsNone(check("gh pr view --json title | jq .title"))

    def test_wc_mid_pipeline(self):
        self.assertIsNone(check("gh issue list | wc -l"))

    def test_multi_stage_pipeline(self):
        self.assertIsNone(check("gh pr list | grep open | head -5"))

    def test_grep_standalone_denied(self):
        self.assertIsNotNone(check("grep foo bar.rs"))

    def test_denied_source_blocks_pipeline(self):
        self.assertIsNotNone(check("cat file | grep foo"))
        self.assertIsNotNone(check("ls | grep foo"))


class TestHeredoc(unittest.TestCase):
    def test_git_commit_heredoc(self):
        self.assertIsNone(check("git commit -m \"$(cat <<'EOF'\nmessage\nEOF\n)\""))

    def test_gh_pr_create_heredoc(self):
        self.assertIsNone(check("gh pr create --body \"$(cat <<'EOF'\nbody text\nEOF\n)\""))

    def test_heredoc_body_with_semicolons(self):
        """Real-world case: commit message body contains ; and English words."""
        cmd = (
            'git commit -m "$(cat <<\'EOF\'\n'
            'feat: fix hook deny response\n'
            '\n'
            '- Fix display; add suppressOutput and systemMessage\n'
            '- Add chmod +x to script (missing execute bit)\n'
            'EOF\n'
            ')"'
        )
        self.assertIsNone(check(cmd))

    def test_heredoc_body_with_parentheses(self):
        """Heredoc body with ) that would truncate _SUBSHELL_RE capture."""
        cmd = (
            'git commit -m "$(cat <<\'EOF\'\n'
            'fix(hook): missing execute bit was silently allowing blocked commands through)\n'
            'EOF\n'
            ')"'
        )
        self.assertIsNone(check(cmd))

    def test_cat_file_still_denied(self):
        self.assertIsNotNone(check("cat file.txt"))


class TestSubshell(unittest.TestCase):
    def test_subshell_cat_standalone(self):
        self.assertIsNotNone(check("$(cat Makefile)"))

    def test_subshell_cat_in_git_arg(self):
        self.assertIsNotNone(check("git commit -m \"$(cat file)\""))

    def test_backtick_grep(self):
        self.assertIsNotNone(check("`grep foo bar`"))

    def test_backtick_cat(self):
        self.assertIsNotNone(check("make build `cat args.txt`"))


class TestProcessSubstitution(unittest.TestCase):
    def test_cat_inside_denied(self):
        self.assertIsNotNone(check("git diff <(cat file1) <(cat file2)"))

    def test_grep_inside_denied(self):
        self.assertIsNotNone(check("git diff <(grep foo bar)"))

    def test_git_show_inside_allowed(self):
        self.assertIsNone(check("git diff <(git show HEAD:src/main.rs) <(git show HEAD~1:src/main.rs)"))


class TestSequential(unittest.TestCase):
    def test_and_denied(self):
        self.assertIsNotNone(check("make build && cat file"))

    def test_semicolon_denied(self):
        self.assertIsNotNone(check("git status; ls"))

    def test_or_denied(self):
        self.assertIsNotNone(check("make check || cargo test"))

    def test_both_allowed(self):
        self.assertIsNone(check("git fetch && make check"))


class TestAdversarial(unittest.TestCase):
    """Patterns an LLM might try as bypass attempts."""

    # --- env-var prefix tricks ---

    def test_env_var_before_denied(self):
        # find_command must skip the assignment and still catch grep
        self.assertIsNotNone(check("FOO=0 grep foo bar"))

    def test_multiple_env_vars_before_denied(self):
        self.assertIsNotNone(check("A=1 B=2 cat file"))

    def test_env_var_before_denied_full_path(self):
        self.assertIsNotNone(check("PATH=/tmp /usr/bin/grep foo bar"))

    # --- subshell spacing and nesting ---

    def test_subshell_with_internal_spaces(self):
        # $( cat file ) — spaces inside the parens
        self.assertIsNotNone(check("$( cat file )"))

    def test_nested_subshell(self):
        # $(echo $(cat file)) — outer echo is denied; inner cat also denied
        self.assertIsNotNone(check("$(echo $(cat file))"))

    def test_subshell_in_pipeline_position(self):
        # mid-pipeline subshell still has its own pipe_pos=0 context internally
        self.assertIsNotNone(check("gh pr list | $(cat file)"))

    def test_subshell_grep_in_pipeline(self):
        # grep inside a subshell mid-pipeline — grep reads a file, not stdin
        self.assertIsNotNone(check("gh pr list | $(grep foo bar.rs)"))

    def test_backtick_in_git_arg(self):
        self.assertIsNotNone(check("git commit -m \"`cat file`\""))

    # --- semicolon tricks ---

    def test_semicolon_leading_subshell(self):
        # ; $(head file) — first segment is empty, second is the subshell
        self.assertIsNotNone(check("; $(head file)"))

    def test_semicolon_then_cat_subshell(self):
        self.assertIsNotNone(check("make check; $(cat Makefile)"))

    def test_semicolon_then_grep(self):
        self.assertIsNotNone(check("git status; grep foo bar"))

    # --- here-string (<<<) should not open the heredoc escape hatch ---

    def test_head_with_herestring_denied(self):
        # <<< is a here-string, not a heredoc — head still reads "from a string"
        # but we don't want this to silently bypass the cat/head block
        # Current behaviour: startswith("<<") matches <<<, so this passes.
        # Document the known behaviour rather than asserting a direction.
        result = check("head -5 <<< 'hello'")
        # here-string is benign (reads a literal string, not a file); allowed is acceptable
        _ = result  # not asserting — just confirming it doesn't raise

    def test_herestring_with_subshell_denied(self):
        # The subshell inside the here-string must still be caught
        self.assertIsNotNone(check("head -5 <<< $(cat /etc/passwd)"))

    # --- || vs | confusion ---

    def test_logical_or_both_checked(self):
        # || splits sequentially; grep on right side is at pipe_pos=0 → denied
        self.assertIsNotNone(check("make check || grep foo bar"))

    # --- path traversal ---

    def test_relative_path_denied(self):
        self.assertIsNotNone(check("./grep foo bar"))
        self.assertIsNotNone(check("../bin/grep foo bar"))

    # --- git with process substitution ---

    def test_git_diff_process_substitution_denied(self):
        self.assertIsNotNone(check("git diff <(cat file1) <(cat file2)"))


class TestQuotedOperators(unittest.TestCase):
    """Operators inside quoted strings must not be treated as shell operators."""

    def test_awk_with_and_in_pattern(self):
        self.assertIsNone(check("make test | awk '/a/ && /b/' | sort"))

    def test_pipe_inside_single_quotes(self):
        self.assertIsNone(check("make test | awk '/a|b/ {print}'"))

    def test_semicolon_inside_single_quotes(self):
        self.assertIsNone(check("make ARGS='a;b;c' test"))

    def test_and_inside_double_quotes(self):
        self.assertIsNone(check('git commit -m "foo && bar"'))

    def test_pipe_inside_double_quotes(self):
        self.assertIsNone(check('git commit -m "a | b"'))

    def test_semicolon_inside_double_quotes(self):
        self.assertIsNone(check('git commit -m "a; b"'))

    def test_unquoted_operators_still_split(self):
        # Ensure real operators outside quotes are still enforced
        self.assertIsNotNone(check("make build && cat file"))
        self.assertIsNotNone(check("make build; ls ."))
        self.assertIsNotNone(check("cat file | grep foo"))


class TestDenyResponse(unittest.TestCase):
    def test_claude_system_message_contains_command(self):
        response = deny_response("claude", "cargo build 2>&1", "Use a make target instead.")
        self.assertIn("systemMessage", response)
        self.assertIn("cargo build 2>&1", response["systemMessage"])

    def test_claude_suppress_output(self):
        response = deny_response("claude", "cargo build 2>&1", "Use a make target instead.")
        self.assertTrue(response.get("suppressOutput"))

    def test_claude_hook_specific_output(self):
        response = deny_response("claude", "cargo build 2>&1", "Use a make target instead.")
        hso = response.get("hookSpecificOutput", {})
        self.assertEqual(hso.get("hookEventName"), "PreToolUse")
        self.assertEqual(hso.get("permissionDecision"), "deny")
        self.assertEqual(hso.get("permissionDecisionReason"), "Use a make target instead.")

    def test_gemini_format(self):
        response = deny_response("gemini", "cargo build 2>&1", "Use a make target instead.")
        self.assertEqual(response.get("decision"), "deny")
        self.assertEqual(response.get("reason"), "Use a make target instead.")
        self.assertNotIn("hookSpecificOutput", response)


if __name__ == "__main__":
    unittest.main(verbosity=2)