diff --git a/truss/templates/control/control/helpers/truss_patch/model_code_patch_applier.py b/truss/templates/control/control/helpers/truss_patch/model_code_patch_applier.py index 78ca3a36e..66cc3b4f2 100644 --- a/truss/templates/control/control/helpers/truss_patch/model_code_patch_applier.py +++ b/truss/templates/control/control/helpers/truss_patch/model_code_patch_applier.py @@ -13,9 +13,33 @@ from truss.templates.control.control.helpers.custom_types import Action, Patch +def _validate_within(base_dir: Path, relative_path: str) -> Path: + """Ensure ``relative_path`` joined onto ``base_dir`` stays inside + ``base_dir``, then return the plain join. + + The patch path comes from the request body of the control server's + ``/control/patch`` endpoint, so a value such as ``../../../etc/cron.d/x`` + (or an absolute path, which ``/`` join would leave unchanged) must not be + allowed to write or delete files outside the target directory. + + Resolved copies are used only for the containment check; the returned + path is ``base_dir / relative_path`` exactly as before this check was + introduced, so downstream file operations (and their log output) are + unchanged for legitimate patches, including when ``base_dir`` sits + behind a symlink. + """ + filepath = base_dir / relative_path + if not filepath.resolve().is_relative_to(base_dir.resolve()): + raise ValueError( + f"Invalid patch path {relative_path!r}: resolves outside the " + f"target directory {base_dir}." + ) + return filepath + + def apply_code_patch(relative_dir: Path, patch: Patch, logger: logging.Logger): logger.debug(f"Applying code patch {patch.to_dict()}") - filepath: Path = relative_dir / patch.path + filepath: Path = _validate_within(relative_dir, patch.path) action = patch.action if action in [Action.ADD, Action.UPDATE]: diff --git a/truss/tests/templates/control/control/helpers/test_model_container_patch_applier.py b/truss/tests/templates/control/control/helpers/test_model_container_patch_applier.py index 2f56e66de..a07786501 100644 --- a/truss/tests/templates/control/control/helpers/test_model_container_patch_applier.py +++ b/truss/tests/templates/control/control/helpers/test_model_container_patch_applier.py @@ -168,6 +168,99 @@ def test_patch_applier_env_var_patch_remove(patch_applier: ModelContainerPatchAp _ = env_var_dict["FOO"] +@pytest.mark.parametrize( + "patch_type, body_cls", + [(PatchType.MODEL_CODE, ModelCodePatch), (PatchType.PACKAGE, PackagePatch)], +) +@pytest.mark.parametrize("action", [Action.ADD, Action.UPDATE]) +def test_patch_applier_rejects_path_traversal_write( + patch_applier: ModelContainerPatchApplier, + truss_container_fs, + patch_type, + body_cls, + action, +): + # A patch path escaping the target directory (e.g. via `..`) must be + # rejected instead of writing outside it (issue #2532). + outside_target = truss_container_fs / "escaped.py" + patch = Patch( + type=patch_type, + body=body_cls(action=action, path="../../escaped.py", content="pwned"), + ) + with pytest.raises(ValueError): + patch_applier(patch, os.environ.copy()) + assert not outside_target.exists() + + +def test_patch_applier_rejects_absolute_path_write( + patch_applier: ModelContainerPatchApplier, truss_container_fs, tmp_path +): + # Joining an absolute path onto the target dir collapses to the absolute + # path, another way to escape; it must be rejected too. + outside_target = tmp_path / "abs_escaped.py" + patch = Patch( + type=PatchType.MODEL_CODE, + body=ModelCodePatch( + action=Action.ADD, path=str(outside_target), content="pwned" + ), + ) + with pytest.raises(ValueError): + patch_applier(patch, os.environ.copy()) + assert not outside_target.exists() + + +def test_patch_applier_rejects_path_traversal_remove( + patch_applier: ModelContainerPatchApplier, truss_container_fs +): + # A REMOVE patch must not be able to delete files outside the target dir. + victim = truss_container_fs / "app" / "config.yaml" + assert victim.exists() + patch = Patch( + type=PatchType.MODEL_CODE, + body=ModelCodePatch(action=Action.REMOVE, path="../config.yaml"), + ) + with pytest.raises(ValueError): + patch_applier(patch, os.environ.copy()) + assert victim.exists() + + +def test_patch_applier_allows_nested_subdirectory_path( + patch_applier: ModelContainerPatchApplier, truss_container_fs +): + # Legitimate nested paths within the target dir must still work. + patch = Patch( + type=PatchType.MODEL_CODE, + body=ModelCodePatch(action=Action.ADD, path="nested/dir/new.py", content="ok"), + ) + patch_applier(patch, os.environ.copy()) + assert ( + truss_container_fs / "app" / "model" / "nested" / "dir" / "new.py" + ).read_text() == "ok" + + +def test_patch_applier_works_through_symlinked_app_dir(truss_container_fs, tmp_path): + # The containment check must not alter behavior when the inference server + # home is reached via a symlink: the patch is applied through the original + # (unresolved) path, exactly as before the check was introduced. + symlinked_app = tmp_path / "app_link" + symlinked_app.symlink_to(truss_container_fs / "app") + applier = ModelContainerPatchApplier(symlinked_app, mock.Mock()) + patch = Patch( + type=PatchType.MODEL_CODE, + body=ModelCodePatch(action=Action.UPDATE, path="model.py", content="updated"), + ) + applier(patch, os.environ.copy()) + assert (truss_container_fs / "app" / "model" / "model.py").read_text() == "updated" + # Traversal is still rejected through the symlink. + evil = Patch( + type=PatchType.MODEL_CODE, + body=ModelCodePatch(action=Action.ADD, path="../../escaped.py", content="x"), + ) + with pytest.raises(ValueError): + applier(evil, os.environ.copy()) + assert not (truss_container_fs / "escaped.py").exists() + + def test_patch_applier_external_data_patch_add( patch_applier: ModelContainerPatchApplier, truss_container_fs ):