diff --git a/ms_agent/permission/shell_validator.py b/ms_agent/permission/shell_validator.py index 506a8d073..e0e23e10f 100644 --- a/ms_agent/permission/shell_validator.py +++ b/ms_agent/permission/shell_validator.py @@ -126,8 +126,12 @@ def check(self, command: str, *, _depth: int = 0) -> SafetyDecision: if not tokens: continue - # 3. Check output redirections on the raw sub-command string - redirect_result = self._check_redirects(sub_cmd) + # 3. Check output redirections on the raw sub-command string. + # Resolve relative targets against the cd-tracked cwd, matching + # how the shell itself resolves them. Note this runs before the + # `cd` below updates _current_cwd, which is also what the shell + # does for a command like `cd foo > log`. + redirect_result = self._check_redirects(sub_cmd, cwd=_current_cwd) if redirect_result.action != 'allow': return redirect_result @@ -325,7 +329,11 @@ def _validate_paths( return SafetyDecision( action='allow', reason=f'{cmd_name}: all paths validated') - def _check_redirects(self, sub_cmd: str) -> SafetyDecision: + def _check_redirects(self, + sub_cmd: str, + *, + cwd: str | None = None) -> SafetyDecision: + effective_cwd = cwd or self._workspace_root for match in _REDIRECT_PATTERN.finditer(sub_cmd): target = match.group(1) if _FD_REDIRECT.match(target): @@ -341,7 +349,7 @@ def _check_redirects(self, sub_cmd: str) -> SafetyDecision: result = validate_path( target, - self._workspace_root, + effective_cwd, self._allowed_dirs, 'create', read_only_dirs=self._read_only_dirs, diff --git a/tests/permission/test_shell_validator.py b/tests/permission/test_shell_validator.py index 2f214b3ab..97151740b 100644 --- a/tests/permission/test_shell_validator.py +++ b/tests/permission/test_shell_validator.py @@ -5,7 +5,8 @@ import pytest -from ms_agent.permission.shell_validator import ShellPathValidator +from ms_agent.permission.shell_validator import (PathSafetyConfig, + ShellPathValidator) @pytest.fixture @@ -99,6 +100,60 @@ def test_redirect_with_variable(self, validator): assert r.action == 'deny' +class TestRedirectCwdTracking: + """Relative redirect targets must resolve against the cd-tracked cwd. + + The shell resolves ``>`` targets against the current working directory, + so a compound command like ``cd sub && echo x > ../f`` writes to + ``/sub/../f``. Validating the target against the workspace root + instead disagrees with what actually happens on disk in both directions. + """ + + @pytest.fixture + def layout(self, tmp_path): + work = tmp_path / 'a' / 'b' / 'work' + work2 = tmp_path / 'a' / 'b' / 'work2' + cache = tmp_path / 'cache' + for p in (work / 'sub', work2, cache): + p.mkdir(parents=True) + allowed = (str(work), str(work2), str(cache)) + validator = ShellPathValidator( + allowed_dirs=list(allowed), + safety_config=PathSafetyConfig( + allowed_directories=allowed, workspace_root=str(work)), + ) + return validator, work, work2, cache + + def test_relative_redirect_after_cd_into_subdir(self, layout): + """``cd work/sub && echo x > ../f`` writes work/f — inside allowed.""" + validator, work, _, _ = layout + r = validator.check(f'cd {work}/sub && echo x > ../f') + assert r.action == 'allow' + + def test_relative_redirect_escaping_after_cd(self, layout): + """``cd cache && echo x > ../work2/f`` writes outside allowed dirs. + + Resolved against the workspace root the target would look like + ``work/../work2/f`` (an allowed directory), but the shell writes to + ``cache/../work2/f``, which is not allowed. + """ + validator, _, _, cache = layout + r = validator.check(f'cd {cache} && echo x > ../work2/f') + assert r.action != 'allow' + + def test_redirect_without_cd_uses_workspace_root(self, layout): + validator, _, _, _ = layout + r = validator.check('echo x > f') + assert r.action == 'allow' + + def test_redirect_matches_argument_path_validation(self, layout): + """Redirects and ordinary path arguments must agree on the cwd.""" + validator, work, _, _ = layout + rm = validator.check(f'cd {work}/sub && rm ../f') + redirect = validator.check(f'cd {work}/sub && echo x > ../f') + assert rm.action == redirect.action == 'allow' + + class TestProcessSubstitution: def test_output_substitution(self, validator): r = validator.check('echo secret > >(tee .git/config)')