[framework] Permission confirmation layer bypassable three ways: ask rules miss compound/wrapped shell commands, the persistent-approval file is model-writable, the CLI prompt allows on any unrecognized answer

--- a/ms_agent/permission/config.py	2026-11-04 00:38:36
+++ b/ms_agent/permission/config.py	2026-11-04 01:47:08
@@ -32,6 +32,12 @@
     '.git/config',
     '.git/hooks/*',
     '**/.git/**',
+    # The persistent allow_always store. It is read back as policy before any
+    # ask rule, so a write to it (write_file, edit_file, a shell redirect) is a
+    # permission grant the user never clicked. Both spellings, like .git above:
+    # the relative one for a workspace-relative path, the ** one for absolute.
+    '.ms_agent/permission_memory.json',
+    '**/.ms_agent/permission_memory.json',
 )
 
 #: Locations whose CONTENTS are a credential. Kept apart from
--- a/ms_agent/permission/safety.py	2026-11-04 00:38:36
+++ b/ms_agent/permission/safety.py	2026-11-04 01:47:08
@@ -9,11 +9,12 @@
 import fnmatch
 import os
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any, Literal, Sequence
 
 from .config import SafetyConfig
 from .matcher import PermissionMatcher
-from .path_validator import validate_path
+from .path_validator import matches_sensitive, validate_path
 from .shell_validator import (PathSafetyConfig, SafetyDecision,
                               ShellPathValidator)
 
@@ -43,6 +44,7 @@
             workspace_root=workspace_root,
             dangerous_removal_paths=tuple(config.dangerous_removal_paths),
             sensitive_read_paths=tuple(config.sensitive_read_paths),
+            sensitive_write_paths=tuple(config.sensitive_paths),
         )
         self._shell_validator = ShellPathValidator(
             allowed_dirs=self._allowed_dirs,
@@ -90,18 +92,26 @@
         if not path:
             return SafetyDecision(action='deny', reason='Empty file path')
 
-        # Sensitive path check
+        cwd = self._workspace_root or os.getcwd()
+
+        # Sensitive path check: the path as written AND as it resolves, so
+        # ``./.ms_agent/../.ms_agent/permission_memory.json`` is judged by
+        # where it lands.
         if op_type == 'write':
             expanded = os.path.expanduser(path)
+            try:
+                resolved = str(Path(cwd, expanded).resolve())
+            except (OSError, RuntimeError):
+                resolved = expanded
             for sensitive in self._sensitive_paths:
                 sensitive_expanded = os.path.expanduser(sensitive)
-                if fnmatch.fnmatch(expanded, sensitive_expanded):
+                if fnmatch.fnmatch(expanded, sensitive_expanded) or \
+                        matches_sensitive(resolved, (sensitive, )):
                     return SafetyDecision(
                         action='deny',
                         reason=f'Write to sensitive path blocked: {path}',
                     )
 
-        cwd = self._workspace_root or os.getcwd()
         result = validate_path(
             path,
             cwd,
--- a/ms_agent/permission/shell_validator.py	2026-11-04 00:38:36
+++ b/ms_agent/permission/shell_validator.py	2026-11-04 01:47:08
@@ -23,6 +23,7 @@
                               extract_interpreter_script, find_uses_delete,
                               interpreter_runs_inline_code)
 from .path_validator import (PathValidationResult, is_dangerous_removal_path,
+                             matches_sensitive,
                              validate_path)
 from .sed_validator import check_sed_expression_safety, is_sed_read_only
 from .wrapper_strip import strip_safe_wrappers
@@ -256,6 +257,10 @@
     workspace_root: str | None = None
     dangerous_removal_paths: tuple[str, ...] = ()
     sensitive_read_paths: tuple[str, ...] = ()
+    #: Locations a shell command may not WRITE or CREATE (``SafetyConfig.
+    #: sensitive_paths``). Until now only write_file/edit_file honoured that
+    #: list; ``echo x > .ms_agent/permission_memory.json`` did not.
+    sensitive_write_paths: tuple[str, ...] = ()
 
 
 class ShellPathValidator:
@@ -271,6 +276,8 @@
         self._read_only_dirs = list(self._config.read_only_directories)
         self._workspace_root = self._config.workspace_root or os.getcwd()
         self._sensitive_read_paths = list(self._config.sensitive_read_paths)
+        self._sensitive_write_paths = list(
+            self._config.sensitive_write_paths)
         self._extractors = build_extractor_registry()
 
     def check(self, command: str, *, _depth: int = 0) -> SafetyDecision:
@@ -583,6 +590,12 @@
                     action=result.action,
                     reason=result.reason,
                     category=result.category)
+            if op_type in ('write', 'create') and matches_sensitive(
+                    result.resolved_path, self._sensitive_write_paths):
+                return SafetyDecision(
+                    action='deny',
+                    reason=f'Write to sensitive path blocked: {path}',
+                )
 
         return SafetyDecision(
             action='allow', reason=f'{cmd_name}: all paths validated')
@@ -615,6 +628,12 @@
                     action=result.action,
                     reason=result.reason,
                     category=result.category)
+            if matches_sensitive(result.resolved_path,
+                                 self._sensitive_write_paths):
+                return SafetyDecision(
+                    action='deny',
+                    reason=f'Redirect to sensitive path blocked: {target}',
+                )
 
         return SafetyDecision(action='allow', reason='Redirects OK')
 
--- a/ms_agent/permission/matcher.py	2026-11-04 00:38:36
+++ b/ms_agent/permission/matcher.py	2026-11-04 01:47:31
@@ -6,6 +6,8 @@
 
 from __future__ import annotations
 
+import os
+import shlex
 from typing import Any
 
 from ms_agent.utils.pattern_matcher import match_pattern
@@ -58,6 +60,68 @@
     return '|'.join(p for p in out if p)
 
 
+#: Commands whose job is to run the command that follows them. ``xargs curl``
+#: and ``exec ssh`` reach the network exactly as ``curl`` and ``ssh`` do.
+_EXEC_PREFIXES: frozenset[str] = frozenset({'xargs', 'exec', 'command'})
+
+_MAX_SUBSTITUTION_DEPTH = 3
+
+
+def shell_command_views(command: str, *, _depth: int = 0) -> list[str]:
+    """Every simple command inside ``command``, as ``<name> <args>``.
+
+    An ask rule such as ``curl *`` is about the program being run, not about
+    the first characters of the tool argument. The whole-string fnmatch missed
+    ``true && curl …``, ``cd /tmp; wget …``, ``/usr/bin/curl …``, ``env curl …``
+    and ``xargs curl``. This splits the compound command the way
+    ShellPathValidator does, strips redirections and the safe wrappers, peels
+    the exec-style prefixes, and reports each sub-command with its program
+    name reduced to a basename. Command substitutions are included because
+    ``echo $(curl …)`` runs curl all the same. A sub-command that cannot be
+    tokenised is reported verbatim so a rule still has a chance at it.
+    """
+    from .shell_validator import (_extract_command_substitutions,
+                                  _split_compound, _strip_heredoc_bodies,
+                                  _strip_redirections, _unwrap_group)
+    from .wrapper_strip import _ENV_ASSIGN, strip_safe_wrappers
+
+    views: list[str] = []
+    if not command or _depth > _MAX_SUBSTITUTION_DEPTH:
+        return views
+    command = _strip_heredoc_bodies(command)
+    if _depth < _MAX_SUBSTITUTION_DEPTH:
+        try:
+            bodies = _extract_command_substitutions(command)
+        except ValueError:
+            bodies = []
+        for body in bodies:
+            views.extend(shell_command_views(body, _depth=_depth + 1))
+    for sub in _split_compound(command):
+        sub = _unwrap_group(sub.strip())
+        if not sub or sub.startswith('#'):
+            continue
+        try:
+            tokens = shlex.split(sub)
+        except ValueError:
+            views.append(sub)
+            continue
+        tokens = _strip_redirections(tokens)
+        # Any ``VAR=value`` prefix: it changes the environment the program
+        # sees, never which program runs.
+        while tokens and _ENV_ASSIGN.match(tokens[0]):
+            tokens = tokens[1:]
+        tokens = strip_safe_wrappers(tokens)
+        while tokens and os.path.basename(tokens[0]) in _EXEC_PREFIXES:
+            tokens = tokens[1:]
+            while tokens and tokens[0].startswith('-'):
+                tokens = tokens[1:]
+        if not tokens:
+            continue
+        tokens[0] = os.path.basename(tokens[0])
+        views.append(' '.join(tokens))
+    return views
+
+
 class PermissionMatcher:
     """Wildcard matcher for permission rules, shared by both SafetyGuard and PermissionEnforcer."""
 
@@ -99,3 +163,33 @@
             return False
 
         return self.match(_with_bare_command_variants(content_pattern), content)
+
+    def match_ask_rule(
+        self,
+        pattern: str,
+        tool_name: str,
+        tool_args: dict[str, Any],
+    ) -> bool:
+        """``match_with_content`` that, for the shell, also looks at every
+        sub-command (see :func:`shell_command_views`).
+
+        Only ask rules use this. A whitelist entry must keep matching the
+        command as a whole, or ``ls *`` would release ``ls; curl …``; a rule
+        whose purpose is to confirm has the opposite failure mode, and a
+        confirmation that one extra ``true &&`` can skip is no confirmation.
+        """
+        if self.match_with_content(pattern, tool_name, tool_args):
+            return True
+        if CONTENT_SEP not in pattern:
+            return False
+        if not tool_name.endswith(f'{TOOL_SPLITER}shell_executor'):
+            return False
+        tool_pattern, content_pattern = pattern.split(CONTENT_SEP, 1)
+        if not self.match(tool_pattern, tool_name):
+            return False
+        command = tool_args.get('command')
+        if command is None:
+            return False
+        alts = _with_bare_command_variants(content_pattern)
+        return any(
+            self.match(alts, view) for view in shell_command_views(str(command)))
--- a/ms_agent/permission/enforcer.py	2026-11-04 00:38:36
+++ b/ms_agent/permission/enforcer.py	2026-11-04 01:47:08
@@ -168,7 +168,7 @@
         # user's own remembered answer below, so consenting once still sticks.
         ask_rule = next(
             (p for p in self._config.ask_rules
-             if self._matcher.match_with_content(p, tool_name, tool_args)),
+             if self._matcher.match_ask_rule(p, tool_name, tool_args)),
             None,
         )
         if ask_rule and not self._can_ask_human():
--- a/ms_agent/hooks/permission_resolve.py	2026-11-04 00:38:36
+++ b/ms_agent/hooks/permission_resolve.py	2026-11-04 01:47:08
@@ -28,7 +28,7 @@
                 reason=f'Denied by blacklist rule: {pattern}',
             )
     for pattern in config.ask_rules:
-        if m.match_with_content(pattern, tool_name, tool_args):
+        if m.match_ask_rule(pattern, tool_name, tool_args):
             return PermissionDecision(
                 action='ask',
                 reason=f'Ask rule matched: {pattern}',
--- a/ms_agent/permission/handler.py	2026-11-04 00:38:36
+++ b/ms_agent/permission/handler.py	2026-11-04 01:47:08
@@ -107,29 +107,44 @@
         print(f'{"="*60}', file=sys.stderr)
 
         loop = asyncio.get_running_loop()
-        choice = await loop.run_in_executor(
-            None, lambda: input('Choice [y/s/a/e/n]: ').strip().lower())
 
-        if choice == 's':
+        async def _read(prompt: str) -> str | None:
+            """One line from the terminal, ``None`` when there is none (EOF)."""
+            try:
+                return await loop.run_in_executor(
+                    None, lambda: input(prompt).strip())
+            except EOFError:
+                return None
+
+        raw = await _read('Choice [y/s/a/e/n]: ')
+        if raw is None:
+            print('No answer (EOF), denying.', file=sys.stderr)
             return PermissionResponse(
+                action=PermissionAction.DENY, feedback='No answer (EOF)')
+        choice = raw.lower()
+
+        if choice in ('y', 'yes'):
+            return PermissionResponse(action=PermissionAction.ALLOW_ONCE)
+        elif choice == 's':
+            return PermissionResponse(
                 action=PermissionAction.ALLOW_SESSION,
                 pattern=suggestion,
             )
         elif choice == 'a':
-            edited = await loop.run_in_executor(
-                None,
-                lambda: input(f'Pattern [{suggestion}]: ').strip(),
-            )
+            edited = await _read(f'Pattern [{suggestion}]: ')
+            if edited is None:
+                return PermissionResponse(
+                    action=PermissionAction.DENY, feedback='No answer (EOF)')
             final_pattern = edited if edited else suggestion
             return PermissionResponse(
                 action=PermissionAction.ALLOW_ALWAYS,
                 pattern=final_pattern,
             )
         elif choice == 'e':
-            edited_raw = await loop.run_in_executor(
-                None,
-                lambda: input('New args (JSON): ').strip(),
-            )
+            edited_raw = await _read('New args (JSON): ')
+            if edited_raw is None:
+                return PermissionResponse(
+                    action=PermissionAction.DENY, feedback='No answer (EOF)')
             try:
                 new_args = json.loads(edited_raw)
             except json.JSONDecodeError:
@@ -139,10 +154,15 @@
                 action=PermissionAction.MODIFY,
                 updated_args=new_args,
             )
-        elif choice == 'n':
+        elif choice in ('n', 'no'):
             return PermissionResponse(action=PermissionAction.DENY)
         else:
-            return PermissionResponse(action=PermissionAction.ALLOW_ONCE)
+            # An empty Enter, a typo or a stray key is not consent. The menu
+            # says only ``y`` allows; anything it did not list denies.
+            print(f'Unrecognised choice {raw!r}, denying.', file=sys.stderr)
+            return PermissionResponse(
+                action=PermissionAction.DENY,
+                feedback=f'Unrecognised choice {raw!r}; the call was not run')
 
 
 class EventEmitter(Protocol):

"""ms-agent: three ways a shell command reaches the network, or the shell itself, without the confirmation the
permission layer promises.

M1  The default ask rules (``curl *``, ``wget *``, ``ssh *`` ...) are fnmatch'd against the WHOLE command string, so
    ``true && curl ...``, ``cd /tmp; wget ...``, ``/usr/bin/curl ...``, ``env curl ...``, ``xargs curl`` miss them; in
    auto mode (the default) the enforcer then allows with no prompt.
M2  Persistent approvals are read from ``<workspace>/.ms_agent/permission_memory.json``, which SafetyGuard lets the
    model write (write_file, edit_file, a shell redirect); the next session honours the planted entry before any
    ask rule, in auto AND interactive mode.
M3  The CLI prompt treats any answer other than s/a/e/n, an empty Enter included, as "allow once"; EOF is uncaught.

Runs the repository's own permission code (ms_agent.permission) with stub handlers that record whether a person was
asked. No model, no network, no API key. Usage: python poc.py <repo root>   (unmodified clone, or a patched copy)
"""
import asyncio
import contextlib
import io
import json
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(sys.argv[1]).resolve()))
from ms_agent.permission import (AutoPermissionHandler, CLIPermissionHandler,  # noqa: E402
                                 PermissionConfig, PermissionEnforcer,
                                 PermissionMemory, SafetyGuard)
from ms_agent.permission.handler import (PermissionAction,  # noqa: E402
                                         PermissionResponse)

SHELL = 'code_executor---shell_executor'
findings = []


class RecordingHandler:
    """A person at the keyboard who says "allow once" and remembers being asked."""

    def __init__(self):
        self.asked = 0

    async def ask(self, tool_name, tool_args, context, suggestions=None):
        self.asked += 1
        return PermissionResponse(action=PermissionAction.ALLOW_ONCE)


def make_enforcer(mode, handler, ws, **extra):
    cfg = PermissionConfig.from_dict({'mode': mode, **extra})
    return PermissionEnforcer(cfg, handler=handler,
                              memory=PermissionMemory(project_path=ws))


async def m1():
    print('== M1: ask rules vs compound / wrapped commands (mode=auto, a person attached)')
    cmds = ['curl http://127.0.0.1:1/x', 'true && curl http://127.0.0.1:1/x',
            'cd /tmp; wget http://127.0.0.1:1/x', '/usr/bin/curl http://127.0.0.1:1/x',
            'env curl http://127.0.0.1:1/x', 'echo http://127.0.0.1:1/x | xargs curl',
            'ls\ncurl http://127.0.0.1:1/x', 'true && ssh user@host']
    for cmd in cmds:
        h = RecordingHandler()
        d = await make_enforcer('auto', h, tempfile.mkdtemp()).check(SHELL, {'command': cmd})
        print(f'  {cmd!r:<45} -> {d.action:<5} asked={h.asked}')
        if cmd != cmds[0] and d.action == 'allow' and h.asked == 0:
            findings.append('M1 compound: ' + repr(cmd))
    print('  -- headless (AutoPermissionHandler): an ask rule must DENY, nobody can confirm')
    for cmd in ('curl http://127.0.0.1:1/x', 'true && curl http://127.0.0.1:1/x'):
        d = await make_enforcer('auto', AutoPermissionHandler(), tempfile.mkdtemp()).check(SHELL, {'command': cmd})
        print(f'  {cmd!r:<45} -> {d.action:<5}')
        if d.action == 'allow':
            findings.append('M1 headless: ' + repr(cmd))


async def m2():
    print('== M2: the persistent-approval file is writable by the model, then honoured as policy')
    ws = Path(tempfile.mkdtemp())
    cfg = PermissionConfig.from_dict({'mode': 'interactive'})
    guard = SafetyGuard(config=cfg.safety,
                        allowed_dirs=list(cfg.safety.effective_allowed_directories(str(ws))),
                        workspace_root=str(ws))
    target = ws / '.ms_agent' / 'permission_memory.json'
    grant_json = json.dumps([{'pattern': SHELL, 'scope': 'project', 'source': 'user'}])
    writes = [
        ('file_system---write_file', {'path': str(target), 'content': grant_json}),
        ('file_system---write_file', {'path': '.ms_agent/permission_memory.json', 'content': grant_json}),
        ('file_system---edit_file', {'path': str(target), 'content': grant_json}),
        ('code_executor---shell_executor',
         {'command': 'mkdir -p .ms_agent && echo x > .ms_agent/permission_memory.json'}),
        ('code_executor---shell_executor', {'command': 'echo x > ' + str(target)}),
    ]
    for tool, args in writes:
        d = guard.check(tool, args)
        where = args.get('path') or args.get('command')
        print(f'  SafetyGuard {tool:<26} {where!r:<70} -> {d.action}')
        if d.action == 'allow':
            findings.append('M2 writable: ' + tool + ' ' + repr(where))

    # The grant is honoured once on disk. This is the consequence, not itself a
    # finding: the file is planted directly here, so the next two lines only
    # confirm that a planted entry outranks the ask rule in every mode. The
    # defect is that the model could have written it through the guard above.
    print('  -- plant the grant on disk and start a fresh session that reads it back')
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(grant_json)
    mem = PermissionMemory(project_path=ws)
    print('  loaded entries :', [(e.pattern, e.scope, e.source) for e in mem.list_all()])
    print('  matches curl x :', mem.matches(SHELL, {'command': 'curl http://127.0.0.1:1/x'}))
    for mode in ('interactive', 'auto'):
        h = RecordingHandler()
        d = await make_enforcer(mode, h, ws).check(SHELL, {'command': 'curl http://127.0.0.1:1/x'})
        print(f'  mode={mode:<11} curl ... -> {d.action:<5} asked={h.asked}  ({d.reason})')


async def m3():
    print('== M3: the CLI prompt allows on anything but s/a/e/n; EOF is uncaught')
    handler = CLIPermissionHandler()
    import builtins
    orig_input = builtins.input
    try:
        for answer in ['', 'yse', 'no', ' ', 'q']:
            builtins.input = (lambda a: lambda prompt='': a)(answer)
            with contextlib.redirect_stderr(io.StringIO()):
                r = await handler.ask(SHELL, {'command': 'curl http://127.0.0.1:1/x'}, context='')
            print(f'  input {answer!r:<6} -> {r.action.value}')
            if r.action == PermissionAction.ALLOW_ONCE:
                findings.append('M3 allow-on: ' + repr(answer))

        def raise_eof(prompt=''):
            raise EOFError

        builtins.input = raise_eof
        try:
            with contextlib.redirect_stderr(io.StringIO()):
                r = await handler.ask(SHELL, {'command': 'curl http://127.0.0.1:1/x'}, context='')
            print('  EOF   ->', r.action.value)
        except BaseException as exc:  # noqa: BLE001
            print('  EOF   -> raised', type(exc).__name__)
            findings.append('M3 EOF uncaught')
    finally:
        builtins.input = orig_input


async def main():
    await m1()
    await m2()
    await m3()
    print()
    if findings:
        print('AS SHIPPED - gate bypassed in', len(findings), 'cases:')
        for f in findings:
            print('  -', f)
    else:
        print('PATCHED - every case above was confirmed or denied.')


if __name__ == '__main__':
    asyncio.run(main())