beam: path_sandbox module with traversal/symlink guards + tests
This commit is contained in:
1 parent
89a992d324
commit
ef9a9a956c
2 files changed
+162
No files matched your search
@@ -0,0 +1,89 @@
|
||||
"""Sandboxed path resolution: keep untrusted relative paths inside a root.
|
||||
|
||||
Guards against path traversal, absolute paths, NUL bytes, backslash smuggling,
|
||||
and symlink escapes (resolved via realpath containment). All public functions
|
||||
raise PathEscapeError on any unsafe input instead of returning an escaping path.
|
||||
|
||||
Public API:
|
||||
PathEscapeError
|
||||
safe_join(root, *parts) -> str
|
||||
safe_resolve(root, rel) -> str
|
||||
is_within(root, target) -> bool
|
||||
realpath_root(root) -> str
|
||||
"""
|
||||
import os
|
||||
|
||||
__all__ = ["PathEscapeError", "safe_join", "safe_resolve", "is_within", "realpath_root"]
|
||||
|
||||
|
||||
class PathEscapeError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def realpath_root(root):
|
||||
if not isinstance(root, str) or root == "":
|
||||
raise PathEscapeError("empty root")
|
||||
r = os.path.realpath(root)
|
||||
if not os.path.isdir(r):
|
||||
raise PathEscapeError("root is not an existing directory: %r" % root)
|
||||
return r
|
||||
|
||||
|
||||
def _check_part(part):
|
||||
if not isinstance(part, str):
|
||||
raise PathEscapeError("path component must be str, got %r" % (type(part),))
|
||||
if part == "":
|
||||
return # empty components are harmless in join
|
||||
if "\x00" in part:
|
||||
raise PathEscapeError("NUL byte in path component: %r" % part)
|
||||
# normalize separators: reject backslashes on all platforms to avoid smuggling
|
||||
if "\\" in part:
|
||||
raise PathEscapeError("backslash in path component: %r" % part)
|
||||
if part.startswith("/"):
|
||||
raise PathEscapeError("absolute path component: %r" % part)
|
||||
# split on / and validate each
|
||||
for seg in part.split("/"):
|
||||
if seg in ("", "."):
|
||||
continue
|
||||
if seg == "..":
|
||||
raise PathEscapeError("traversal component '..' in %r" % part)
|
||||
if seg.startswith("/"):
|
||||
raise PathEscapeError("absolute segment in %r" % part)
|
||||
|
||||
|
||||
def safe_join(root, *parts):
|
||||
rr = realpath_root(root)
|
||||
for p in parts:
|
||||
_check_part(p)
|
||||
# build candidate without resolving yet
|
||||
candidate = os.path.join(rr, *parts) if parts else rr
|
||||
# normalize (collapses . and redundant slashes) but does not resolve symlinks
|
||||
candidate = os.path.normpath(candidate)
|
||||
# containment check on the normalized (pre-symlink) form
|
||||
if candidate != rr and not candidate.startswith(rr + os.sep):
|
||||
raise PathEscapeError("joined path escapes root: %r" % candidate)
|
||||
# now resolve symlinks and re-check containment
|
||||
resolved = os.path.realpath(candidate)
|
||||
if resolved != rr and not resolved.startswith(rr + os.sep):
|
||||
raise PathEscapeError("symlink escape: %r -> %r" % (candidate, resolved))
|
||||
return resolved
|
||||
|
||||
|
||||
def safe_resolve(root, rel):
|
||||
if isinstance(rel, (list, tuple)):
|
||||
return safe_join(root, *rel)
|
||||
return safe_join(root, rel)
|
||||
|
||||
|
||||
def is_within(root, target):
|
||||
try:
|
||||
rr = realpath_root(root)
|
||||
except PathEscapeError:
|
||||
return False
|
||||
if not isinstance(target, str) or target == "":
|
||||
return False
|
||||
try:
|
||||
rt = os.path.realpath(target)
|
||||
except (OSError, ValueError):
|
||||
return False
|
||||
return rt == rr or rt.startswith(rr + os.sep)
|
||||
@@ -0,0 +1,73 @@
|
||||
import os, tempfile, unittest
|
||||
from path_sandbox import PathEscapeError, safe_join, safe_resolve, is_within, realpath_root
|
||||
|
||||
|
||||
class TestPathSandbox(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.d = tempfile.mkdtemp()
|
||||
self.root = os.path.join(self.d, "root")
|
||||
os.makedirs(self.root)
|
||||
self.rr = os.path.realpath(self.root)
|
||||
# a file inside
|
||||
with open(os.path.join(self.root, "a.txt"), "w") as f:
|
||||
f.write("x")
|
||||
# subdir
|
||||
os.makedirs(os.path.join(self.root, "sub"))
|
||||
# symlink escaping root
|
||||
os.symlink("/etc", os.path.join(self.root, "escape"))
|
||||
|
||||
def test_basic_join(self):
|
||||
self.assertEqual(safe_join(self.root, "a.txt"), os.path.join(self.rr, "a.txt"))
|
||||
|
||||
def test_nested(self):
|
||||
self.assertEqual(safe_join(self.root, "sub", "b.txt"), os.path.join(self.rr, "sub", "b.txt"))
|
||||
|
||||
def test_traversal_rejected(self):
|
||||
with self.assertRaises(PathEscapeError):
|
||||
safe_join(self.root, "..", "etc", "passwd")
|
||||
|
||||
def test_absolute_rejected(self):
|
||||
with self.assertRaises(PathEscapeError):
|
||||
safe_join(self.root, "/etc/passwd")
|
||||
|
||||
def test_backslash_rejected(self):
|
||||
with self.assertRaises(PathEscapeError):
|
||||
safe_join(self.root, "a\\..\\..\\etc")
|
||||
|
||||
def test_nul_rejected(self):
|
||||
with self.assertRaises(PathEscapeError):
|
||||
safe_join(self.root, "a\x00b")
|
||||
|
||||
def test_symlink_escape_rejected(self):
|
||||
with self.assertRaises(PathEscapeError):
|
||||
safe_join(self.root, "escape", "passwd")
|
||||
|
||||
def test_empty_root_error(self):
|
||||
with self.assertRaises(PathEscapeError):
|
||||
realpath_root("")
|
||||
|
||||
def test_nonexistent_root_error(self):
|
||||
with self.assertRaises(PathEscapeError):
|
||||
realpath_root(os.path.join(self.d, "nope"))
|
||||
|
||||
def test_is_within_true(self):
|
||||
self.assertTrue(is_within(self.root, os.path.join(self.root, "a.txt")))
|
||||
|
||||
def test_is_within_false(self):
|
||||
self.assertFalse(is_within(self.root, "/etc/passwd"))
|
||||
|
||||
def test_safe_resolve_list(self):
|
||||
self.assertEqual(safe_resolve(self.root, ["sub", "x"]), os.path.join(self.rr, "sub", "x"))
|
||||
|
||||
def test_dot_components_ok(self):
|
||||
# single "." components collapse
|
||||
p = safe_join(self.root, ".", "a.txt")
|
||||
self.assertEqual(p, os.path.join(self.rr, "a.txt"))
|
||||
|
||||
def test_double_slash_ok(self):
|
||||
p = safe_join(self.root, "sub//b.txt")
|
||||
self.assertEqual(p, os.path.join(self.rr, "sub", "b.txt"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user