merge: guard the CLI command, not the word

hooks/tea-guard.sh

@@ -27,6 +27,27 @@ Rules:
   - --login "$GITEA_LOGIN", pin found ............... REWRITE to the pin, allow
   - --login "$GITEA_LOGIN", no pin .................. BLOCK (run /tea:auth)
 
+"A `tea` command" means the shell would RUN `tea`, not that the string contains
+the word. The guard used to ask the second question — a substring search over
+the whole command line — and in a repository whose subject *is* the CLI that is
+a different question with the same answer far too often: an issue title, a
+commit message, `grep -rn " tea " docs/` and `echo tea` were all blocked, with
+a message telling the operator to add `--login` to `git commit`. Worse, the
+advice was unfollowable: the only way past the guard was to reword the prose.
+
+So the command is tokenized (heredoc bodies dropped, line continuations
+folded, backticks and newlines treated as boundaries) and only words in
+*command position* count — the first word, and the first word after `;`, `&&`,
+`||`, `|`, `&`, `(`, `)`, `{`, `}`, past any VAR=value assignments and prefix
+words like `env`/`sudo`/`xargs`. Quoting is what saves the prose: a title or a
+`-m` message is one token, and one token is never a command. Compound commands
+stay guarded segment by segment, substitutions included, and every `tea` in the
+line is checked — not just the first.
+
+If the line cannot be tokenized at all (unbalanced quotes), the old substring
+test decides. That direction fails closed: it over-matches, and over-matching
+blocks.
+
 Output protocol: exit 0 + JSON {hookSpecificOutput:{updatedInput,...}} to
 rewrite; exit 2 + stderr to block.
 """
@@ -46,6 +67,131 @@ except Exception:
 
 PLACEHOLDERS = {"$GITEA_LOGIN", "${GITEA_LOGIN}"}
 
+# Operators after which the next word is a command again.
+SEPARATORS = {";", ";;", "&", "&&", "|", "|&", "||", "(", ")", "{", "}"}
+# Words that stand in front of a command without being one.
+TRANSPARENT = {"env", "command", "exec", "nohup", "time", "sudo", "xargs",
+               "if", "then", "else", "elif", "while", "until", "do", "!"}
+
+ASSIGNMENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=")
+REDIRECT = re.compile(r"^\d*[<>]+&?\d*-?$")
+HEREDOC = re.compile(r"<<-?\s*(['\"]?)([A-Za-z_][A-Za-z0-9_]*)\1")
+# A login flag and its value, in the ORIGINAL text — this is what gets
+# rewritten, so it works on the raw string rather than on tokens.
+LOGIN_FLAG = re.compile(r"(--login|(?<![\w-])-l)(\s+|=)(\S+)")
+# The pre-tokenizer test, kept for the one case tokenizing cannot serve.
+LOOKS_LIKE_TEA = re.compile(r"(^|[;&|(]|\s)tea(\s|$)")
+
+NO_LOGIN = ('every `tea` command must include --login "$GITEA_LOGIN" '
+            '(the guard substitutes the operator-pinned login). '
+            'Run /tea:auth if no login is pinned.')
+
+
+def named_login(raw):
+    return ('do not name the login yourself (got `%s`). Write exactly '
+            '--login "$GITEA_LOGIN"; the guard replaces it with the login '
+            'the operator pinned via /tea:auth. This prevents acting under '
+            'the wrong identity.' % raw)
+
+
+def unquote(value):
+    for q in ('"', "'"):
+        if len(value) >= 2 and value[0] == q and value[-1] == q:
+            return value[1:-1]
+    return value
+
+
+def strip_heredocs(cmd):
+    """Drop heredoc bodies. They are data the shell feeds to a command, not
+    commands — and a commit message quoting a raw `tea api` call is exactly the
+    thing that used to be unwritable."""
+    lines, kept, i = cmd.split("\n"), [], 0
+    while i < len(lines):
+        line = lines[i]
+        kept.append(line)
+        i += 1
+        for m in HEREDOC.finditer(line):
+            delim, dash = m.group(2), m.group(0).startswith("<<-")
+            while i < len(lines):
+                probe = lines[i].strip() if dash else lines[i].rstrip()
+                i += 1
+                if probe == delim:
+                    break
+    return "\n".join(kept)
+
+
+def shell_words(cmd):
+    """Tokens, with operators as tokens of their own and quotes honored.
+
+    Backticks and newlines become separators before tokenizing: shlex knows
+    neither, and both start a command. Inside quotes that substitution is
+    harmless — the token still spans the quotes, and a token is never a
+    command."""
+    text = strip_heredocs(cmd)
+    text = re.sub(r"\\\n", " ", text)
+    text = text.replace("`", " ; ").replace("\n", " ; ")
+    lex = shlex.shlex(text, posix=True, punctuation_chars=True)
+    lex.whitespace_split = True
+    return list(lex)
+
+
+def tea_invocations(words):
+    """The argument list of every `tea` the shell would actually run."""
+    found, current, expect, skip = [], None, True, False
+    for w in words:
+        if skip:
+            skip = False
+            continue
+        if REDIRECT.match(w):
+            skip = True          # the target of a redirection is not a command
+            continue
+        if w in SEPARATORS:
+            current, expect = None, True
+            continue
+        if expect:
+            if ASSIGNMENT.match(w) or w in TRANSPARENT:
+                continue
+            expect = False
+            if w.rsplit("/", 1)[-1] == "tea":
+                current = []
+                found.append(current)
+            continue
+        if current is not None:
+            current.append(w)
+    return found
+
+
+def is_meta(args):
+    """Login enumeration and `--version`/`--help`: no identity is used, and
+    /tea:auth needs `tea logins list` while no pin exists yet."""
+    if not args:
+        return False
+    if args[0] in ("--version", "-v", "--help", "-h", "help"):
+        return True
+    return args[0] in ("logins", "login") and len(args) > 1 \
+        and args[1] in ("list", "ls")
+
+
+def login_value(args):
+    """The login as written, or None if the flag is absent."""
+    for i, a in enumerate(args):
+        if a in ("--login", "-l"):
+            return args[i + 1] if i + 1 < len(args) else ""
+        if a.startswith("--login=") or a.startswith("-l="):
+            return a.split("=", 1)[1]
+    return None
+
+
+def substitute(cmd, login):
+    """Every placeholder login in the line, replaced by the pin. Every one:
+    a command may run `tea` twice, and half a rewrite leaves the second call
+    with an unset variable and no login at all."""
+    def repl(m):
+        if unquote(m.group(3)) in PLACEHOLDERS:
+            return m.group(1) + m.group(2) + shlex.quote(login)
+        return m.group(0)
+    return LOGIN_FLAG.sub(repl, cmd)
+
 
 def block(msg):
     sys.stderr.write("tea-guard: BLOCKED — " + msg + "\n")
@@ -81,34 +227,37 @@ def main():
     tool_input = payload.get("tool_input") or {}
     cmd = tool_input.get("command") or ""
 
-    # Not a `tea` invocation → not our concern.
+    try:
-    if not re.search(r'(^|[;&|(]|\s)tea(\s|$)', cmd):
+        runs = tea_invocations(shell_words(cmd))
-        allow_passthrough()
+    except ValueError:
+        # Unbalanced quotes: what the shell would run is not knowable here.
+        # Fall back to the substring test — it over-matches, and over-matching
+        # blocks rather than lets an unpinned call through.
+        runs = None
 
-    # Whitelist: login enumeration + meta. No identity is used; /tea:auth
+    if runs is None:
-    # needs `tea logins list` while no pin exists yet.
+        if not LOOKS_LIKE_TEA.search(cmd):
-    if re.search(r'tea\s+(logins\s+(list|ls)|--version|-v|--help|help)(\s|$)', cmd):
             allow_passthrough()
-
+        m = LOGIN_FLAG.search(cmd)
-    # Locate --login / -l and its value (logins never contain spaces).
-    m = re.search(r'(--login|(?<![\w-])-l)(\s+|=)(\S+)', cmd)
         if not m:
-        block('every `tea` command must include --login "$GITEA_LOGIN" '
+            block(NO_LOGIN)
-              '(the guard substitutes the operator-pinned login). '
+        if unquote(m.group(3)) not in PLACEHOLDERS:
-              'Run /tea:auth if no login is pinned.')
+            block(named_login(m.group(3)))
-
+    else:
-    raw_val = m.group(3)
+        # The word appears but nothing runs it → not our concern. This is the
-    inner = raw_val
+        # branch that lets prose about the CLI be written at all.
-    for q in ('"', "'"):
+        if not runs:
-        if len(inner) >= 2 and inner[0] == q and inner[-1] == q:
+            allow_passthrough()
-            inner = inner[1:-1]
+        for args in runs:
-            break
+            if is_meta(args):
-
+                continue
-    if inner not in PLACEHOLDERS:
+            raw = login_value(args)
-        block('do not name the login yourself (got `%s`). Write exactly '
+            if raw is None:
-              '--login "$GITEA_LOGIN"; the guard replaces it with the login '
+                block(NO_LOGIN)
-              'the operator pinned via /tea:auth. This prevents acting under '
+            if unquote(raw) not in PLACEHOLDERS:
-              'the wrong identity.' % raw_val)
+                block(named_login(raw))
+        if all(is_meta(args) for args in runs):
+            allow_passthrough()
 
     if pin is None:
         block('cannot import skills/auth/scripts/pin.py, so the pinned login '
@@ -123,8 +272,7 @@ def main():
               '.claude/settings.local.json env.GITEA_LOGIN). The guard reads '
               'the file at call time, so it takes effect with no restart.')
 
-    new_cmd = cmd[:m.start(3)] + shlex.quote(login) + cmd[m.end(3):]
+    rewrite(tool_input, substitute(cmd, login),
-    rewrite(tool_input, new_cmd,
             'tea-guard: resolved --login -> %s (pinned in %s)' % (login, src))
 
 

tests/test_guard_word_boundary.py

@@ -0,0 +1,211 @@
+#!/usr/bin/env python3
+"""
+What the guard guards: `tea` the command, not `tea` the word.
+
+    python3 -m unittest discover -s tests -v
+
+The bug these tests hold down: the guard asked whether the string contained
+`tea` surrounded by whitespace, so in a repository *about* the CLI it blocked
+prose. An issue title, a commit message quoting a raw call, `grep -rn " tea "`
+and `echo tea` were all refused, with a message telling the operator to add
+`--login` to `git commit`. The advice could not be followed — the only way
+past was to reword the sentence.
+
+Two lines are held at once here, and neither may move without the other: the
+four false positives pass, and every shape that really runs the CLI — after
+`&&`, after a pipe, in a subshell, in a substitution, twice in one line — is
+still blocked or still rewritten. A test that only proved the first would be
+satisfied by deleting the guard.
+
+No network and no `tea` binary: the hook is pure decision-making, so the
+fixture is a directory with a pin in it and a JSON payload on stdin.
+"""
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+GUARD = os.path.join(REPO, "hooks", "tea-guard.sh")
+
+sys.path.insert(0, os.path.join(REPO, "skills", "auth", "scripts"))
+import pin  # noqa: E402
+
+LOGIN = "fixture/user"
+
+ALLOW, BLOCK, REWRITE = "allow", "block", "rewrite"
+
+
+class GuardCase(unittest.TestCase):
+    """One temp project with one pinned login; the hook run as the harness
+    runs it."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory(prefix="tea-guard-")
+        self.root = os.path.realpath(self._tmp.name)
+        self.addCleanup(self._tmp.cleanup)
+        path = pin.settings_path(self.root)
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "w") as f:
+            f.write(json.dumps({"env": {pin.ENV_KEY: LOGIN}}))
+
+    def run_guard(self, cmd):
+        env = dict(os.environ)
+        env.pop("PYTHONPATH", None)
+        env[pin.PROJECT_DIR_ENV] = self.root
+        p = subprocess.run([sys.executable, GUARD],
+                           input=json.dumps({"tool_input": {"command": cmd},
+                                             "cwd": self.root}),
+                           cwd=self.root, env=env,
+                           capture_output=True, text=True)
+        return p
+
+    def verdict(self, cmd):
+        p = self.run_guard(cmd)
+        if p.returncode == 2:
+            return BLOCK, p.stderr
+        self.assertEqual(p.returncode, 0, p.stderr)
+        if not p.stdout.strip():
+            return ALLOW, ""
+        got = json.loads(p.stdout)["hookSpecificOutput"]["updatedInput"]["command"]
+        return REWRITE, got
+
+    def assertVerdict(self, cmd, expected):
+        kind, detail = self.verdict(cmd)
+        self.assertEqual(kind, expected,
+                         "%r → %s (%s)" % (cmd, kind, detail.strip()))
+        return detail
+
+
+# --------------------------------------------------------------------------
+# the four false positives, verbatim from the report
+# --------------------------------------------------------------------------
+
+class TestProseAboutTheCliRuns(GuardCase):
+
+    def test_an_issue_title_may_name_the_command(self):
+        self.assertVerdict(
+            'python3 skills/issue/scripts/issue_new.py --type bug '
+            '--title "Warn that tea pulls create needs the repo checkout" '
+            '--label comp/use --severity low', ALLOW)
+
+    def test_a_commit_message_may_quote_a_raw_call(self):
+        self.assertVerdict(
+            "git add -A && git commit -F- <<'EOF'\n"
+            "feat: close issues through a script\n"
+            "\n"
+            "Единственным способом сменить state был сырой вызов\n"
+            "tea api -X PATCH ... repos/OWNER/REPO/issues/N\n"
+            "EOF", ALLOW)
+
+    def test_a_one_line_commit_message_may_too(self):
+        self.assertVerdict('git commit -m "route it through tea api"', ALLOW)
+
+    def test_searching_the_repository_for_the_word(self):
+        for cmd in ('grep -rn " tea " docs/',
+                    'grep -rn "tea api" skills/',
+                    'echo tea'):
+            self.assertVerdict(cmd, ALLOW)
+
+    def test_the_word_as_a_bare_argument_is_still_an_argument(self):
+        """`echo tea` was the smallest case in the report; these are the same
+        shape with the word in other argument positions."""
+        for cmd in ('ls tea', 'cat notes/tea', 'python3 x.py tea api'):
+            self.assertVerdict(cmd, ALLOW)
+
+
+# --------------------------------------------------------------------------
+# and the real thing is still guarded
+# --------------------------------------------------------------------------
+
+class TestRealInvocationsStayGuarded(GuardCase):
+
+    def test_a_bare_call_without_a_login_is_blocked(self):
+        detail = self.assertVerdict("tea issues list", BLOCK)
+        self.assertIn("--login", detail)
+
+    def test_the_placeholder_is_rewritten_to_the_pin(self):
+        got = self.assertVerdict(
+            'tea issues list --login "$GITEA_LOGIN" --state open', REWRITE)
+        self.assertIn(LOGIN, got)
+        self.assertNotIn("GITEA_LOGIN", got)
+
+    def test_a_login_named_by_hand_is_blocked(self):
+        detail = self.assertVerdict("tea issues list --login somebody", BLOCK)
+        self.assertIn("do not name the login", detail)
+
+    def test_another_variable_is_not_the_placeholder(self):
+        self.assertVerdict('tea issues list --login "$OTHER"', BLOCK)
+
+    def test_compound_commands_are_read_segment_by_segment(self):
+        for cmd in ('cd /tmp && tea issues list',
+                    'echo x | tea api -X GET repos/x/y',
+                    '( tea issues list )',
+                    'cd /tmp; tea issues list',
+                    'FOO=1 tea issues list',
+                    'sudo tea issues list',
+                    'xargs tea issues list'):
+            self.assertVerdict(cmd, BLOCK)
+
+    def test_substitutions_are_read_too(self):
+        for cmd in ('echo $(tea whoami)',
+                    'x=$(tea whoami)',
+                    'echo `tea whoami`'):
+            self.assertVerdict(cmd, BLOCK)
+
+    def test_a_guarded_call_beside_prose_that_mentions_the_word(self):
+        """The two halves of the bug in one line: the guard must ignore the
+        argument and still catch the call."""
+        self.assertVerdict(
+            'git commit -m "route it through tea api" && tea issues list',
+            BLOCK)
+
+    def test_an_absolute_path_to_the_binary_is_the_binary(self):
+        self.assertVerdict("/usr/local/bin/tea issues list", BLOCK)
+
+    def test_every_call_in_the_line_is_rewritten(self):
+        """A half-rewritten line leaves the second call with an unset variable
+        and therefore no login at all."""
+        got = self.assertVerdict(
+            'tea issues list --login "$GITEA_LOGIN" && '
+            'tea pulls list --login "$GITEA_LOGIN"', REWRITE)
+        self.assertEqual(got.count(LOGIN), 2)
+        self.assertNotIn("GITEA_LOGIN", got)
+
+    def test_a_second_unguarded_call_is_not_covered_by_the_first(self):
+        self.assertVerdict(
+            'tea issues list --login "$GITEA_LOGIN" && tea pulls list', BLOCK)
+
+    def test_prose_naming_the_whitelisted_form_does_not_launder_a_call(self):
+        """`tea logins list` is allowed because it uses no identity. Quoting
+        that phrase must not turn the call beside it into a whitelisted one."""
+        self.assertVerdict(
+            'echo "run tea logins list first" && tea issues list', BLOCK)
+
+
+class TestTheWhitelistStillApplies(GuardCase):
+
+    def test_login_enumeration_needs_no_pin(self):
+        for cmd in ("tea logins list", "tea logins ls",
+                    "tea --version", "tea --help"):
+            self.assertVerdict(cmd, ALLOW)
+
+    def test_a_whitelisted_call_next_to_a_guarded_one_does_not_excuse_it(self):
+        self.assertVerdict("tea logins list && tea issues list", BLOCK)
+
+
+class TestUnparseableLinesFailClosed(GuardCase):
+    """An unbalanced quote means the shell's reading and ours may differ. The
+    old substring test decides — it over-matches, and over-matching blocks."""
+
+    def test_an_unterminated_quote_around_a_call_still_blocks(self):
+        self.assertVerdict('tea issues list --state "open', BLOCK)
+
+    def test_an_unterminated_quote_with_no_call_is_still_allowed(self):
+        self.assertVerdict('echo "unterminated', ALLOW)
+
+
+if __name__ == "__main__":
+    unittest.main()