test_workspace_security.py
7.07 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
"""U-036: real unittest coverage for workspace_security.py -- adapted from
graphloop's test_workspace_security.py (2026-08-02). Module logic is
unchanged from graphloop ("adapted verbatim", see workspace_security.py's
own docstring); only the import path changed.
Run: python3 -m unittest discover -s supervisor/isolation -p "test_*.py" -v
"""
import os
import shutil
import sys
import tempfile
import unittest
from supervisor.isolation import workspace_security as ws
class ValidateArtifactsTests(unittest.TestCase):
def setUp(self):
self.workspace = tempfile.mkdtemp(prefix="wssec-workspace-")
self.addCleanup(shutil.rmtree, self.workspace, ignore_errors=True)
self.outside_dir = tempfile.mkdtemp(prefix="wssec-outside-")
self.addCleanup(shutil.rmtree, self.outside_dir, ignore_errors=True)
def _write(self, relative_path, content="content"):
full_path = os.path.join(self.workspace, relative_path)
os.makedirs(os.path.dirname(full_path), exist_ok=True)
with open(full_path, "w", encoding="utf-8") as handle:
handle.write(content)
return full_path
def test_accepts_and_normalizes_plain_relative_artifacts(self):
self._write("output.txt")
self._write(os.path.join("nested", "dir", "file.txt"))
result = ws.validate_artifacts(["output.txt", "nested/dir/file.txt"], self.workspace)
self.assertEqual(result, ["output.txt", "nested/dir/file.txt"])
def test_normalizes_redundant_leading_dot_slash(self):
self._write("output.txt")
result = ws.validate_artifacts(["./output.txt"], self.workspace)
self.assertEqual(result, ["output.txt"])
def test_rejects_non_string_item(self):
self._write("output.txt")
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_artifacts(["output.txt", 42], self.workspace)
def test_rejects_non_list_artifacts_argument(self):
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_artifacts("output.txt", self.workspace)
def test_rejects_absolute_path_artifact(self):
absolute = self._write("output.txt")
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_artifacts([absolute], self.workspace)
def test_rejects_dot_dot_traversal_even_when_target_exists(self):
self._write("output.txt")
os.makedirs(os.path.join(self.workspace, "sibling"), exist_ok=True)
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_artifacts(["sibling/../output.txt"], self.workspace)
def test_rejects_dot_dot_traversal_escaping_workspace(self):
secret = os.path.join(self.outside_dir, "secret.txt")
with open(secret, "w", encoding="utf-8") as handle:
handle.write("secret")
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_artifacts(
[os.path.join("..", os.path.basename(self.outside_dir), "secret.txt")],
self.workspace,
)
def test_rejects_missing_artifact(self):
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_artifacts(["does_not_exist.txt"], self.workspace)
def test_rejects_symlink_artifact(self):
real_file = self._write("real.txt")
symlink_path = os.path.join(self.workspace, "link.txt")
os.symlink(real_file, symlink_path)
self.addCleanup(lambda: os.path.exists(symlink_path) and os.remove(symlink_path))
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_artifacts(["link.txt"], self.workspace)
def test_rejects_artifact_resolving_outside_workspace_via_symlinked_directory(self):
secret = os.path.join(self.outside_dir, "secret.txt")
with open(secret, "w", encoding="utf-8") as handle:
handle.write("secret")
symlinked_dir = os.path.join(self.workspace, "escape")
os.symlink(self.outside_dir, symlinked_dir, target_is_directory=True)
self.addCleanup(lambda: os.path.exists(symlinked_dir) and os.remove(symlinked_dir))
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_artifacts(["escape/secret.txt"], self.workspace)
def test_rejects_empty_string_artifact(self):
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_artifacts([""], self.workspace)
class ValidateSubprocessCommandTests(unittest.TestCase):
def setUp(self):
self.workspace = tempfile.mkdtemp(prefix="wssec-cmd-workspace-")
self.addCleanup(shutil.rmtree, self.workspace, ignore_errors=True)
self.adapter_script = os.path.join(self.workspace, "existing_adapter.py")
with open(self.adapter_script, "w", encoding="utf-8") as handle:
handle.write("print('ok')\n")
def test_accepts_python_executable_with_existing_script(self):
result = ws.validate_subprocess_command([sys.executable, self.adapter_script])
self.assertEqual(result, [sys.executable, self.adapter_script])
def test_rejects_empty_command(self):
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_subprocess_command([])
def test_rejects_non_list_command(self):
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_subprocess_command("{} {}".format(sys.executable, self.adapter_script))
def test_rejects_non_string_argument(self):
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_subprocess_command([sys.executable, self.adapter_script, 7])
def test_rejects_shell_metacharacters_in_arguments(self):
dangerous_args = [
"foo; rm -rf /", "foo && rm -rf /", "foo | rm -rf /",
"foo > /etc/passwd", "foo `rm -rf /`", "foo $(rm -rf /)",
]
for dangerous_arg in dangerous_args:
with self.subTest(dangerous_arg=dangerous_arg):
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_subprocess_command([sys.executable, self.adapter_script, dangerous_arg])
def test_rejects_bare_shell_invocation(self):
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_subprocess_command(["/bin/sh", "-c", "echo hi"])
def test_rejects_script_argument_with_disallowed_extension(self):
non_script = os.path.join(self.workspace, "not_a_script.txt")
with open(non_script, "w", encoding="utf-8") as handle:
handle.write("not python")
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_subprocess_command([sys.executable, non_script])
def test_rejects_script_argument_that_does_not_exist(self):
missing_script = os.path.join(self.workspace, "missing_adapter.py")
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_subprocess_command([sys.executable, missing_script])
def test_rejects_newline_in_argument(self):
with self.assertRaises(ws.WorkspaceSecurityError):
ws.validate_subprocess_command([sys.executable, self.adapter_script, "foo\nbar"])
if __name__ == "__main__":
unittest.main()