omx(team): auto-checkpoint worker-1 [1]

tests/test_security_scope.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+from src.models import PermissionDenial
+from src.path_scope import WorkspacePathScope, extract_path_candidates
+from src.permissions import ToolPermissionContext
+from src.query_engine import QueryEnginePort
+from src.tools import execute_tool
+
+
+class WorkspacePathScopeTests(unittest.TestCase):
+    def test_direct_parent_escape_is_denied(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            workspace = Path(tmp) / 'workspace'
+            workspace.mkdir()
+            decision = WorkspacePathScope.from_root(workspace).validate_payload('cat ../secret.txt')
+            self.assertFalse(decision.allowed)
+            self.assertIn('outside workspace scope', decision.reason)
+
+    def test_issue_3007_symlink_escape_is_denied(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            workspace = root / 'workspace'
+            outside = root / 'outside'
+            workspace.mkdir()
+            outside.mkdir()
+            (outside / 'secret.txt').write_text('secret')
+            link = workspace / 'linked-outside'
+            link.symlink_to(outside, target_is_directory=True)
+
+            decision = WorkspacePathScope.from_root(workspace).validate_payload('cat linked-outside/secret.txt')
+
+            self.assertFalse(decision.allowed)
+            self.assertIn(str(outside.resolve()), decision.resolved or '')
+
+    def test_glob_expansion_must_stay_inside_workspace(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            workspace = root / 'workspace'
+            outside = root / 'outside'
+            workspace.mkdir()
+            outside.mkdir()
+            (outside / 'secret.txt').write_text('secret')
+
+            decision = WorkspacePathScope.from_root(workspace).validate_payload(f'cat {outside}/*.txt')
+
+            self.assertFalse(decision.allowed)
+            self.assertEqual(str((outside / 'secret.txt').resolve()), decision.resolved)
+
+    def test_shell_environment_expansion_is_validated(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            workspace = root / 'workspace'
+            outside = root / 'outside'
+            workspace.mkdir()
+            outside.mkdir()
+            previous = os.environ.get('CLAW_SCOPE_OUTSIDE')
+            os.environ['CLAW_SCOPE_OUTSIDE'] = str(outside)
+            try:
+                self.assertEqual((f'{outside}/secret.txt',), extract_path_candidates('cat $CLAW_SCOPE_OUTSIDE/secret.txt'))
+                decision = WorkspacePathScope.from_root(workspace).validate_payload('cat $CLAW_SCOPE_OUTSIDE/secret.txt')
+            finally:
+                if previous is None:
+                    os.environ.pop('CLAW_SCOPE_OUTSIDE', None)
+                else:
+                    os.environ['CLAW_SCOPE_OUTSIDE'] = previous
+
+            self.assertFalse(decision.allowed)
+            self.assertIn(str(outside.resolve()), decision.resolved or '')
+
+    def test_explicit_worktree_roots_are_allowed(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            workspace = root / 'workspace'
+            worktree = root / 'worktree'
+            workspace.mkdir()
+            worktree.mkdir()
+            (worktree / 'file.txt').write_text('ok')
+
+            decision = WorkspacePathScope.from_roots((workspace, worktree)).validate_payload(f'cat {worktree}/file.txt')
+
+            self.assertTrue(decision.allowed, decision.reason)
+
+    def test_windows_absolute_paths_are_denied_for_posix_workspace(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            workspace = Path(tmp) / 'workspace'
+            workspace.mkdir()
+
+            drive_decision = WorkspacePathScope.from_root(workspace).validate_payload(r'type C:\Users\other\secret.txt')
+            unc_decision = WorkspacePathScope.from_root(workspace).validate_payload(r'type \\server\share\secret.txt')
+
+            self.assertFalse(drive_decision.allowed)
+            self.assertIn('windows absolute path', drive_decision.reason)
+            self.assertFalse(unc_decision.allowed)
+            self.assertIn('windows absolute path', unc_decision.reason)
+
+    def test_file_and_shell_tools_use_workspace_scope_context(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            workspace = root / 'workspace'
+            outside = root / 'outside'
+            workspace.mkdir()
+            outside.mkdir()
+            context = ToolPermissionContext.from_iterables(workspace_root=workspace, cwd=workspace)
+
+            file_result = execute_tool('FileReadTool', f'{outside}/secret.txt', permission_context=context)
+            shell_result = execute_tool('BashTool', f'cat {outside}/secret.txt', permission_context=context)
+            inside_result = execute_tool('FileReadTool', './allowed.txt', permission_context=context)
+
+            self.assertFalse(file_result.handled)
+            self.assertIn('Permission denied', file_result.message)
+            self.assertFalse(shell_result.handled)
+            self.assertIn('Permission denied', shell_result.message)
+            self.assertTrue(inside_result.handled)
+
+    def test_permission_denial_stream_events_expose_status_and_reason(self) -> None:
+        engine = QueryEnginePort.from_workspace()
+        denial = PermissionDenial('BashTool', 'path resolves outside workspace scope')
+
+        events = list(engine.stream_submit_message('cat ../secret.txt', matched_tools=('BashTool',), denied_tools=(denial,)))
+        permission_event = next(event for event in events if event['type'] == 'permission_denial')
+        result = engine.submit_message('cat ../secret.txt', matched_tools=('BashTool',), denied_tools=(denial,))
+
+        self.assertEqual('blocked', permission_event['denials'][0]['status'])
+        self.assertEqual('path resolves outside workspace scope', permission_event['denials'][0]['reason'])
+        self.assertIn('status=blocked', result.output)
+        self.assertIn('path resolves outside workspace scope', result.output)
+
+
+if __name__ == '__main__':
+    unittest.main()