diff --git a/openspace/cloud/client.py b/openspace/cloud/client.py index fb0792f..7bcd3b1 100644 --- a/openspace/cloud/client.py +++ b/openspace/cloud/client.py @@ -340,7 +340,11 @@ class OpenSpaceClient: record_data = self.fetch_record(skill_id) skill_name = record_data.get("name", skill_id) - skill_dir = target_dir / skill_name + if "/" in skill_name or "\\" in skill_name or skill_name.startswith("."): + skill_name = skill_id + skill_dir = (target_dir / skill_name).resolve() + if not skill_dir.is_relative_to(target_dir.resolve()): + raise CloudError(f"Skill name {skill_name!r} escapes target directory") # Check if already exists locally if skill_dir.exists() and (skill_dir / SKILL_FILENAME).exists(): @@ -401,6 +405,7 @@ class OpenSpaceClient: def _extract_zip(zip_data: bytes, target_dir: Path) -> List[str]: """Extract zip bytes to target directory with path traversal protection.""" extracted: List[str] = [] + resolved_target = target_dir.resolve() try: with zipfile.ZipFile(io.BytesIO(zip_data)) as zf: for info in zf.infolist(): @@ -409,7 +414,9 @@ class OpenSpaceClient: clean_name = Path(info.filename).as_posix() if clean_name.startswith("..") or clean_name.startswith("/"): continue - target_path = target_dir / clean_name + target_path = (target_dir / clean_name).resolve() + if not target_path.is_relative_to(resolved_target): + continue target_path.parent.mkdir(parents=True, exist_ok=True) target_path.write_bytes(zf.read(info)) extracted.append(clean_name) diff --git a/tests/test_zip_path_traversal.py b/tests/test_zip_path_traversal.py new file mode 100644 index 0000000..275176b --- /dev/null +++ b/tests/test_zip_path_traversal.py @@ -0,0 +1,89 @@ +"""Regression tests for zip extraction and import_skill path traversal.""" + +import io +import zipfile +from pathlib import Path +from unittest.mock import patch, MagicMock + +import pytest + + +def _make_zip(entries: dict[str, bytes]) -> bytes: + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + for name, data in entries.items(): + zf.writestr(name, data) + return buf.getvalue() + + +def _get_extract_zip(): + import importlib + mod = importlib.import_module("openspace.cloud.client") + return mod.OpenSpaceClient._extract_zip + + +class TestExtractZip: + def test_normal_file_extracted(self, tmp_path): + extract = _get_extract_zip() + zip_data = _make_zip({"hello.txt": b"world"}) + result = extract(zip_data, tmp_path) + assert "hello.txt" in result + assert (tmp_path / "hello.txt").read_bytes() == b"world" + + def test_dotdot_prefix_blocked(self, tmp_path): + extract = _get_extract_zip() + zip_data = _make_zip({"../escape.txt": b"bad"}) + result = extract(zip_data, tmp_path) + assert result == [] + + def test_nested_traversal_blocked(self, tmp_path): + """The real bug: nested/../../escape.txt bypassed the old startswith check.""" + extract = _get_extract_zip() + zip_data = _make_zip({"nested/../../escape.txt": b"bad"}) + result = extract(zip_data, tmp_path) + assert result == [] + assert not (tmp_path.parent / "escape.txt").exists() + + def test_absolute_path_blocked(self, tmp_path): + extract = _get_extract_zip() + zip_data = _make_zip({"/etc/passwd": b"bad"}) + result = extract(zip_data, tmp_path) + assert result == [] + + +class TestImportSkillNameTraversal: + def test_malicious_record_name_sanitized(self, tmp_path): + from openspace.cloud.client import OpenSpaceClient + + client = OpenSpaceClient.__new__(OpenSpaceClient) + target_dir = tmp_path / "skills" + target_dir.mkdir() + + malicious_name = "../../escapedir" + skill_id = "safe_skill_id" + zip_data = _make_zip({"SKILL.md": b"---\nname: test\n---\ncontent"}) + + with patch.object(client, "fetch_record", return_value={"name": malicious_name}), \ + patch.object(client, "download_artifact", return_value=zip_data): + result = client.import_skill(skill_id, target_dir) + + assert result["status"] == "success" + resolved = Path(result["local_path"]).resolve() + assert resolved.is_relative_to(target_dir.resolve()) + assert not (tmp_path.parent / "escapedir").exists() + + def test_normal_record_name_works(self, tmp_path): + from openspace.cloud.client import OpenSpaceClient + + client = OpenSpaceClient.__new__(OpenSpaceClient) + target_dir = tmp_path / "skills" + target_dir.mkdir() + + zip_data = _make_zip({"SKILL.md": b"---\nname: test\n---\ncontent"}) + + with patch.object(client, "fetch_record", return_value={"name": "my-skill"}), \ + patch.object(client, "download_artifact", return_value=zip_data): + result = client.import_skill("some_id", target_dir) + + assert result["status"] == "success" + assert (target_dir / "my-skill" / "SKILL.md").exists()