Harden package paths and storage writes
This commit is contained in:
@@ -1,7 +1,7 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
import json
|
import json
|
||||||
import re
|
import re
|
||||||
from pathlib import Path
|
from pathlib import Path, PureWindowsPath
|
||||||
from beeos.core.models import PackageManifest
|
from beeos.core.models import PackageManifest
|
||||||
|
|
||||||
VALID_TYPES = {"app", "script", "script_pack", "watchface", "theme", "plugin"}
|
VALID_TYPES = {"app", "script", "script_pack", "watchface", "theme", "plugin"}
|
||||||
@@ -13,6 +13,21 @@ def validate_package_id(value: str) -> None:
|
|||||||
if not re.match(r"^[a-z0-9][a-z0-9_.-]+$", value):
|
if not re.match(r"^[a-z0-9][a-z0-9_.-]+$", value):
|
||||||
raise ManifestError("Package id must use lowercase letters, numbers, dots, hyphens, or underscores.")
|
raise ManifestError("Package id must use lowercase letters, numbers, dots, hyphens, or underscores.")
|
||||||
|
|
||||||
|
def validate_entry_path(value: str) -> None:
|
||||||
|
"""Reject manifest entries that can escape the package directory."""
|
||||||
|
if not isinstance(value, str) or not value.strip():
|
||||||
|
raise ManifestError("Entry must be a non-empty relative path.")
|
||||||
|
|
||||||
|
normalized = value.replace("\\", "/")
|
||||||
|
windows_path = PureWindowsPath(value)
|
||||||
|
if (
|
||||||
|
Path(normalized).is_absolute()
|
||||||
|
or windows_path.is_absolute()
|
||||||
|
or windows_path.drive
|
||||||
|
or any(part == ".." for part in normalized.split("/"))
|
||||||
|
):
|
||||||
|
raise ManifestError("Entry must be a relative path inside the package directory.")
|
||||||
|
|
||||||
def load_manifest(path: Path) -> PackageManifest:
|
def load_manifest(path: Path) -> PackageManifest:
|
||||||
data = json.loads(path.read_text(encoding="utf-8"))
|
data = json.loads(path.read_text(encoding="utf-8"))
|
||||||
required = ["id", "name", "type", "version", "author", "description", "entry"]
|
required = ["id", "name", "type", "version", "author", "description", "entry"]
|
||||||
@@ -21,6 +36,7 @@ def load_manifest(path: Path) -> PackageManifest:
|
|||||||
raise ManifestError(f"Missing required manifest field: {field}")
|
raise ManifestError(f"Missing required manifest field: {field}")
|
||||||
|
|
||||||
validate_package_id(data["id"])
|
validate_package_id(data["id"])
|
||||||
|
validate_entry_path(data["entry"])
|
||||||
if data["type"] not in VALID_TYPES:
|
if data["type"] not in VALID_TYPES:
|
||||||
raise ManifestError(f"Invalid package type: {data['type']}")
|
raise ManifestError(f"Invalid package type: {data['type']}")
|
||||||
|
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import importlib.util
|
|||||||
import inspect
|
import inspect
|
||||||
import io
|
import io
|
||||||
import json
|
import json
|
||||||
|
import os
|
||||||
import sys
|
import sys
|
||||||
import traceback
|
import traceback
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
@@ -88,7 +89,10 @@ class BeeStorageAPI:
|
|||||||
return {}
|
return {}
|
||||||
|
|
||||||
def _save(self, data: dict[str, Any]) -> None:
|
def _save(self, data: dict[str, Any]) -> None:
|
||||||
self.storage_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
|
payload = json.dumps(data, indent=2)
|
||||||
|
temporary_file = self.storage_file.with_name(self.storage_file.name + ".tmp")
|
||||||
|
temporary_file.write_text(payload, encoding="utf-8")
|
||||||
|
os.replace(temporary_file, self.storage_file)
|
||||||
|
|
||||||
def get(self, key: str, default: Any = None) -> Any:
|
def get(self, key: str, default: Any = None) -> Any:
|
||||||
return self._load().get(key, default)
|
return self._load().get(key, default)
|
||||||
|
|||||||
@@ -0,0 +1,50 @@
|
|||||||
|
import json
|
||||||
|
import tempfile
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from beeos.core.package_manifest import ManifestError, load_manifest
|
||||||
|
|
||||||
|
|
||||||
|
BASE_MANIFEST = {
|
||||||
|
"id": "local.test.app",
|
||||||
|
"name": "Test App",
|
||||||
|
"type": "app",
|
||||||
|
"version": "1.0.0",
|
||||||
|
"author": "BeeOS",
|
||||||
|
"description": "Manifest validation test package",
|
||||||
|
"entry": "app.py",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class PackageManifestTests(unittest.TestCase):
|
||||||
|
def load_with_entry(self, entry: str):
|
||||||
|
data = {**BASE_MANIFEST, "entry": entry}
|
||||||
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||||||
|
path = Path(temp_dir) / "manifest.json"
|
||||||
|
path.write_text(json.dumps(data), encoding="utf-8")
|
||||||
|
return load_manifest(path)
|
||||||
|
|
||||||
|
def test_accepts_relative_entry(self):
|
||||||
|
manifest = self.load_with_entry("src/app.py")
|
||||||
|
self.assertEqual(manifest.entry, "src/app.py")
|
||||||
|
|
||||||
|
def test_rejects_parent_directory_entry(self):
|
||||||
|
with self.assertRaisesRegex(ManifestError, "relative path"):
|
||||||
|
self.load_with_entry("../outside.py")
|
||||||
|
|
||||||
|
def test_rejects_windows_absolute_entry(self):
|
||||||
|
with self.assertRaisesRegex(ManifestError, "relative path"):
|
||||||
|
self.load_with_entry(r"C:\\outside.py")
|
||||||
|
|
||||||
|
def test_rejects_windows_parent_directory_entry(self):
|
||||||
|
with self.assertRaisesRegex(ManifestError, "relative path"):
|
||||||
|
self.load_with_entry(r"..\outside.py")
|
||||||
|
|
||||||
|
def test_rejects_empty_entry(self):
|
||||||
|
with self.assertRaisesRegex(ManifestError, "non-empty"):
|
||||||
|
self.load_with_entry(" ")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
@@ -0,0 +1,33 @@
|
|||||||
|
import json
|
||||||
|
import tempfile
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from beeos.core.runtime_api import BeeStorageAPI
|
||||||
|
|
||||||
|
|
||||||
|
class BeeStorageAPITests(unittest.TestCase):
|
||||||
|
def test_set_persists_json_and_leaves_no_temporary_file(self):
|
||||||
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||||||
|
storage_file = Path(temp_dir) / "storage.json"
|
||||||
|
storage = BeeStorageAPI(storage_file)
|
||||||
|
|
||||||
|
storage.set("launches", 3)
|
||||||
|
|
||||||
|
self.assertEqual(storage.get("launches"), 3)
|
||||||
|
self.assertEqual(json.loads(storage_file.read_text(encoding="utf-8")), {"launches": 3})
|
||||||
|
self.assertFalse(storage_file.with_name("storage.json.tmp").exists())
|
||||||
|
|
||||||
|
def test_delete_removes_key_without_corrupting_other_values(self):
|
||||||
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||||||
|
storage = BeeStorageAPI(Path(temp_dir) / "storage.json")
|
||||||
|
storage.set("launches", 3)
|
||||||
|
storage.set("theme", "yellow")
|
||||||
|
|
||||||
|
storage.delete("launches")
|
||||||
|
|
||||||
|
self.assertEqual(storage.all(), {"theme": "yellow"})
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user