workspace_security.py
5.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
"""Validation helpers that stand between an untrusted process (a Coder node,
a Validator node, or a candidate deploy instance) and the things that trust
it: the artifact paths it reports back, and the argv used to launch it as a
subprocess.
Both validators raise WorkspaceSecurityError on anything suspicious rather
than silently coercing or ignoring it.
Adapted verbatim from graphloop's workspace_security.py (2026-07-30) --
no functional changes. graphloop's own Phase-10 self-evolution run hit a
real case this class of check is meant to catch (a worker writing files
outside the isolated tree it was supposed to be confined to); keeping this
module's logic unchanged preserves that validated behavior.
"""
from __future__ import annotations
import os
from typing import Any, List
class WorkspaceSecurityError(Exception):
"""Raised by validate_artifacts/validate_subprocess_command on rejection."""
# Shell metacharacters/sequences that must never appear in a subprocess
# argument, since these commands are meant to run with shell=False and any
# of these would only matter if something later reinterpreted them via a
# shell.
_SHELL_METACHARACTERS = (";", "&&", "|", ">", "`", "$(")
def _normalize_relative_posix_parts(artifact: Any) -> List[str]:
if not isinstance(artifact, str):
raise WorkspaceSecurityError(f"artifact path is not a string: {artifact!r}")
if artifact == "":
raise WorkspaceSecurityError("artifact path is empty")
if os.path.isabs(artifact):
raise WorkspaceSecurityError(f"artifact path is absolute: {artifact!r}")
posix_path = artifact.replace(os.sep, "/")
if os.altsep:
posix_path = posix_path.replace(os.altsep, "/")
parts = [part for part in posix_path.split("/") if part not in ("", ".")]
if not parts:
raise WorkspaceSecurityError(f"artifact path is empty: {artifact!r}")
if any(part == ".." for part in parts):
raise WorkspaceSecurityError(
f"artifact path contains a parent traversal segment: {artifact!r}"
)
return parts
def validate_artifacts(artifacts: Any, workspace: str) -> List[str]:
"""Validate and normalize artifact paths reported by an untrusted
child process, ensuring every one lives safely inside `workspace`."""
if not isinstance(artifacts, (list, tuple)):
raise WorkspaceSecurityError(
f"artifacts must be a list of strings, got {type(artifacts).__name__}"
)
workspace_real = os.path.realpath(workspace)
normalized: List[str] = []
for artifact in artifacts:
parts = _normalize_relative_posix_parts(artifact)
current = workspace
for part in parts:
current = os.path.join(current, part)
if os.path.islink(current):
raise WorkspaceSecurityError(
f"artifact path contains a symlink component: {artifact!r}"
)
if not os.path.exists(current):
raise WorkspaceSecurityError(
f"artifact does not exist under workspace: {artifact!r}"
)
real_target = os.path.realpath(current)
if os.path.commonpath([workspace_real, real_target]) != workspace_real:
raise WorkspaceSecurityError(
f"artifact resolves outside workspace: {artifact!r}"
)
normalized.append("/".join(parts))
return normalized
def validate_subprocess_command(command: Any) -> List[str]:
"""Validate an argv list about to be passed to subprocess.run(shell=False).
Accepts the canonical [python_executable, existing_script.py, ...] shape
only, with no shell metacharacters in any argument.
"""
if not isinstance(command, (list, tuple)):
raise WorkspaceSecurityError(
f"command must be a list of strings, got {type(command).__name__}"
)
if len(command) == 0:
raise WorkspaceSecurityError("command must not be empty")
for argument in command:
if not isinstance(argument, str):
raise WorkspaceSecurityError(
f"command argument is not a string: {argument!r}"
)
if "\n" in argument or "\r" in argument:
raise WorkspaceSecurityError(
f"command argument contains a newline: {argument!r}"
)
for metacharacter in _SHELL_METACHARACTERS:
if metacharacter in argument:
raise WorkspaceSecurityError(
f"command argument contains shell metacharacter "
f"{metacharacter!r}: {argument!r}"
)
executable = command[0]
if not os.path.isfile(executable):
raise WorkspaceSecurityError(
f"command executable does not exist: {executable!r}"
)
if len(command) < 2:
raise WorkspaceSecurityError(
f"command is missing a script argument: {list(command)!r}"
)
script = command[1]
if not script.endswith(".py") or not os.path.isfile(script):
raise WorkspaceSecurityError(
f"command script argument is not an existing .py file: {script!r}"
)
return list(command)