test: add path safety, validation, and JSON deserialize regression coverage

Libraries/PyKotor/src/pykotor/tools/archive_serializer.py

@@ -175,9 +175,9 @@ def _dict_to_erf(data: dict[str, Any]) -> bytes:
         payload = r["data"]
         raw = deserialize_embedded_resource_payload(enc, payload, restype)
         erf.set_data(ResRef(resref), restype, raw)
-    out = BytesIO()
+    out: bytearray = bytearray()
     write_erf(erf, out)
-    return out.getvalue()
+    return bytes(out)
 
 
 def _dict_to_rim(data: dict[str, Any]) -> bytes:
@@ -189,9 +189,9 @@ def _dict_to_rim(data: dict[str, Any]) -> bytes:
         payload = r["data"]
         raw = deserialize_embedded_resource_payload(enc, payload, restype)
         rim.set_data(ResRef(resref), restype, raw)
-    out = BytesIO()
+    out: bytearray = bytearray()
     write_rim(rim, out)
-    return out.getvalue()
+    return bytes(out)
 
 
 def _dict_to_bif(data: dict[str, Any]) -> bytes:

Libraries/PyKotor/src/pykotor/tools/resource_json.py

@@ -778,39 +778,39 @@ def deserialize_embedded_resource_payload(
         return base64.b64decode(payload)
     if encoding == "gff_json":
         gff = read_gff(
-            BytesIO(json.dumps(payload, ensure_ascii=False).encode("utf-8")),
+            json.dumps(payload, ensure_ascii=False).encode("utf-8"),
             file_format=ToolsetFormat.GFF_JSON,
         )
         output = bytearray()
         write_gff(gff, output, file_format=ResourceType.GFF)
         return bytes(output)
     if encoding == "tlk_json":
         tlk = read_tlk(
-            BytesIO(json.dumps(payload, ensure_ascii=False).encode("utf-8")),
+            json.dumps(payload, ensure_ascii=False).encode("utf-8"),
             file_format=ToolsetFormat.TLK_JSON,
         )
         output = bytearray()
         write_tlk(tlk, output, file_format=ResourceType.TLK)
         return bytes(output)
     if encoding == "2da_json":
         twoda = read_2da(
-            BytesIO(json.dumps(payload, ensure_ascii=False).encode("utf-8")),
+            json.dumps(payload, ensure_ascii=False).encode("utf-8"),
             file_format=ToolsetFormat.TwoDA_JSON,
         )
         output = bytearray()
         write_2da(twoda, output, file_format=ResourceType.TwoDA)
         return bytes(output)
     if encoding == "lip_json":
         lip = read_lip(
-            BytesIO(json.dumps(payload, ensure_ascii=False).encode("utf-8")),
+            json.dumps(payload, ensure_ascii=False).encode("utf-8"),
             file_format=ToolsetFormat.LIP_JSON,
         )
         output = bytearray()
         write_lip(lip, output, file_format=ResourceType.LIP)
         return bytes(output)
     if encoding == "ssf_json":
         ssf = read_ssf(
-            BytesIO(json.dumps(payload, ensure_ascii=False).encode("utf-8")),
+            json.dumps(payload, ensure_ascii=False).encode("utf-8"),
             file_format=ToolsetFormat.SSF_JSON,
         )
         output = bytearray()

Libraries/PyKotor/tests/cli/test_json_commands.py

@@ -151,6 +151,37 @@ def test_cmd_get_can_extract_from_folder_source(tmp_path: Path) -> None:
     output_path.unlink()
 
 
+def test_cmd_get_rejects_output_path_outside_cwd(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    workdir = tmp_path / "work"
+    workdir.mkdir()
+    source_dir = workdir / "source"
+    source_dir.mkdir()
+    (source_dir / "notes.txt").write_text("folder source", encoding="utf-8")
+
+    monkeypatch.chdir(workdir)
+    escape_path = workdir.parent / "escaped.txt"
+
+    args = Namespace(
+        resref="notes.txt",
+        path=str(source_dir),
+        game=None,
+        path_index=0,
+        source=None,
+        order=None,
+        output=str(escape_path),
+        format="binary",
+    )
+
+    with caplog.at_level(logging.ERROR):
+        assert cmd_get(args, logging.getLogger("test_cmd_get_path_safety")) == 1
+    assert "Output path rejected" in caplog.text
+    assert not escape_path.exists()
+
+
 def test_cmd_find_can_search_archive_source(
     tmp_path: Path, caplog: pytest.LogCaptureFixture
 ) -> None:

Libraries/PyKotor/tests/resource/formats/test_archive_serializer_regressions.py

@@ -14,7 +14,12 @@
 from pykotor.resource.formats.gff.gff_data import GFF, GFFFieldType
 from pykotor.resource.formats.gff.io_gff_json import GFFJSONWriter
 from pykotor.resource.type import ResourceType
-from pykotor.tools.archive_serializer import _resource_bytes_to_plaintext, archive_to_dict
+from pykotor.resource.formats.erf import read_erf
+from pykotor.tools.archive_serializer import (
+    _resource_bytes_to_plaintext,
+    archive_to_dict,
+    dict_to_archive,
+)
 
 
 def test_archive_to_dict_mod_extension_uses_erf_path() -> None:
@@ -67,3 +72,33 @@ def test_read_gff_accepts_bytes_for_plaintext_pipeline() -> None:
     _tag, ftype, value = next(iter(loaded.root))
     assert ftype == GFFFieldType.ResRef
     assert str(value) == "abc"
+
+
+def test_erf_base64_roundtrip_preserves_gff_resource() -> None:
+    """archive_to_dict -> dict_to_archive must preserve ERF resources (base64 encoding)."""
+    gff = GFF()
+    gff.root.set_uint32("Count", 7)
+    original_gff = bytes_gff(gff)
+
+    erf = ERF(ERFType.ERF)
+    erf.set_data(ResRef("testitem"), ResourceType.UTI, original_gff)
+    with tempfile.TemporaryDirectory() as tmp:
+        erf_path = Path(tmp) / "test.erf"
+        write_erf(erf, erf_path)
+        archive_dict = archive_to_dict(erf_path, embed_plaintext=False)
+
+    assert archive_dict["format"] == "erf"
+    assert len(archive_dict["resources"]) == 1
+    assert archive_dict["resources"][0]["data_encoding"] == "base64"
+
+    rebuilt_bytes, extension = dict_to_archive(archive_dict)
+    assert extension == "erf"
+
+    rebuilt_erf = read_erf(rebuilt_bytes)
+    resources = {
+        (str(res.resref).lower(), res.restype): bytes(res.data)
+        for res in rebuilt_erf
+    }
+    key = (str(ResRef("testitem")).lower(), ResourceType.UTI)
+    assert key in resources
+    assert resources[key] == original_gff

Libraries/PyKotor/tests/test_path_safety.py

@@ -0,0 +1,110 @@
+"""Unit tests for extract/write path safety (path_safety.py).
+
+Guards against path traversal and writes outside allowed bases for CLI get and MCP extract.
+"""
+
+from __future__ import annotations
+
+import os
+
+from pathlib import Path
+
+import pytest
+
+from pykotor.tools.path_safety import (
+    MAX_PATH_LENGTH,
+    get_extract_base,
+    resolve_and_validate_under_base,
+    validate_extract_output_path,
+)
+
+
+def test_resolve_accepts_path_under_base(tmp_path: Path) -> None:
+    base = tmp_path / "extract"
+    base.mkdir()
+    target = base / "nested" / "out.txt"
+
+    resolved = resolve_and_validate_under_base(target, base)
+
+    assert resolved == target.resolve()
+    assert resolved.relative_to(base.resolve())
+
+
+def test_resolve_accepts_nonexistent_output_under_base(tmp_path: Path) -> None:
+    base = tmp_path / "extract"
+    base.mkdir()
+    target = base / "new" / "resource.tlk"
+
+    resolved = resolve_and_validate_under_base(target, base, allow_nonexistent=True)
+
+    assert resolved == target.resolve()
+
+
+def test_resolve_rejects_path_outside_base(tmp_path: Path) -> None:
+    base = tmp_path / "allowed"
+    base.mkdir()
+    outside = tmp_path.parent / "escape.txt"
+
+    with pytest.raises(ValueError, match="outside allowed base"):
+        resolve_and_validate_under_base(outside, base)
+
+
+def test_resolve_rejects_traversal_via_parent_segments(tmp_path: Path) -> None:
+    base = tmp_path / "allowed"
+    base.mkdir()
+    traversal = base / ".." / ".." / "etc" / "passwd"
+
+    with pytest.raises(ValueError, match="outside allowed base"):
+        resolve_and_validate_under_base(traversal, base)
+
+
+def test_resolve_rejects_overlong_path(tmp_path: Path) -> None:
+    base = tmp_path / "allowed"
+    base.mkdir()
+    long_name = "a" * (MAX_PATH_LENGTH + 1)
+    long_path = base / long_name
+
+    with pytest.raises(ValueError, match="Path length exceeds maximum"):
+        resolve_and_validate_under_base(long_path, base)
+
+
+def test_resolve_strict_requires_existing_path(tmp_path: Path) -> None:
+    base = tmp_path / "allowed"
+    base.mkdir()
+    missing = base / "does_not_exist.txt"
+
+    with pytest.raises(FileNotFoundError):
+        resolve_and_validate_under_base(missing, base, allow_nonexistent=False)
+
+
+def test_get_extract_base_uses_env_when_set(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    extract_dir = tmp_path / "custom_extract"
+    extract_dir.mkdir()
+    monkeypatch.setenv("PYKOTOR_EXTRACT_DIR", str(extract_dir))
+
+    assert get_extract_base() == extract_dir.resolve()
+
+
+def test_get_extract_base_falls_back_to_cwd(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("PYKOTOR_EXTRACT_DIR", raising=False)
+    assert get_extract_base() == Path.cwd()
+
+
+def test_validate_extract_output_path_uses_provided_base(tmp_path: Path) -> None:
+    base = tmp_path / "base"
+    base.mkdir()
+    target = base / "out" / "file.gff"
+
+    resolved = validate_extract_output_path(target, base=base)
+
+    assert resolved == target.resolve()
+
+
+def test_validate_extract_output_path_rejects_escape(tmp_path: Path) -> None:
+    base = tmp_path / "base"
+    base.mkdir()
+
+    with pytest.raises(ValueError, match="outside allowed base"):
+        validate_extract_output_path(tmp_path.parent / "outside.txt", base=base)

Libraries/PyKotor/tests/test_resource_json_deserialize.py

@@ -0,0 +1,71 @@
+"""Unit tests for embedded resource JSON deserialization (resource_json.py)."""
+
+from __future__ import annotations
+
+import base64
+
+import pytest
+
+from pykotor.resource.formats.gff import bytes_gff, read_gff
+from pykotor.resource.formats.gff.gff_data import GFF, GFFFieldType
+from pykotor.resource.type import ResourceType
+from pykotor.tools.resource_json import (
+    deserialize_embedded_resource_payload,
+    direct_json_document_to_resource_bytes,
+)
+
+
+def test_deserialize_base64_roundtrip() -> None:
+    payload = b"hello archive"
+    encoded = base64.b64encode(payload).decode("ascii")
+
+    result = deserialize_embedded_resource_payload("base64", encoded, ResourceType.TXT)
+
+    assert result == payload
+
+
+def test_deserialize_base64_rejects_non_string_payload() -> None:
+    with pytest.raises(ValueError, match="Base64 payload must be a string"):
+        deserialize_embedded_resource_payload("base64", 123, ResourceType.TXT)
+
+
+def test_deserialize_gff_json_roundtrip() -> None:
+    import json
+
+    from pykotor.resource.formats.gff.io_gff_json import GFFJSONWriter
+
+    gff = GFF()
+    gff.root.set_uint32("Value", 99)
+    out = bytearray()
+    GFFJSONWriter(gff, out).write()
+    payload = json.loads(bytes(out).decode("utf-8"))
+
+    result = deserialize_embedded_resource_payload("gff_json", payload, ResourceType.GFF)
+    loaded = read_gff(result)
+    _label, ftype, value = next(iter(loaded.root))
+    assert ftype == GFFFieldType.UInt32
+    assert value == 99
+
+
+def test_deserialize_unknown_encoding_raises() -> None:
+    with pytest.raises(ValueError, match="Unknown embedded JSON encoding"):
+        deserialize_embedded_resource_payload("not_a_codec", "data", ResourceType.TXT)
+
+
+def test_direct_json_document_binary_wrapper() -> None:
+    payload = b"\xde\xad\xbe\xef"
+    document = {
+        "format": "binary",
+        "restype": "TXT",
+        "encoding": "base64",
+        "data_base64": base64.b64encode(payload).decode("ascii"),
+    }
+
+    result = direct_json_document_to_resource_bytes(document, ResourceType.TXT)
+
+    assert result == payload
+
+
+def test_direct_json_document_rejects_unsupported_format() -> None:
+    with pytest.raises(ValueError, match="does not use a direct-wrapper format"):
+        direct_json_document_to_resource_bytes({"format": "unknown"}, ResourceType.TXT)