fix(security): harden zip extraction and import_skill against path traversal

- Add resolve() + is_relative_to() check in _extract_zip() to block
  nested traversal entries like nested/../../escape.txt
- Sanitize server-provided skill name in import_skill() to prevent
  directory escape via malicious record metadata
- Add 6 regression tests covering both attack vectors

Closes #17

Co-authored-by: LeftX <xzq-xu@users.noreply.github.com>

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)

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()