diff --git a/tests/test_app_config.py b/tests/test_app_config.py index 94ecdf2..4e357b6 100644 --- a/tests/test_app_config.py +++ b/tests/test_app_config.py @@ -153,6 +153,78 @@ def test_missing_source_file(self): has_missing = any("not found" in m.lower() or "missing" in m.lower() or "does not exist" in m.lower() for m in all_messages) self.assertTrue(has_missing, f"expected missing file message in: {all_messages}") + def test_source_outside_app_is_rejected(self): + yaml = """ +metadata: + name: External Source + bundle_id: org.test.external + foreground: + process: + cmd: [python, app.py] +steps: + - name: Copy + type: files + files: + - source: ../secret.txt + destination: ./secret.txt +""" + with tempfile.TemporaryDirectory() as tmp: + Path(tmp, "secret.txt").write_text("secret") + app_dir = self._make_app(tmp, yaml, {"app.py": "pass"}) + valid, _config, _app_type, _warnings, errors = validate_app_dir(app_dir) + + self.assertFalse(valid) + self.assertTrue(any("source file must stay within" in error.lower() for error in errors)) + + def test_external_icon_is_rejected(self): + yaml = """ +metadata: + name: External Icon + bundle_id: org.test.external-icon + icon_file: ../icon.png + foreground: + process: + cmd: [python, app.py] +""" + with tempfile.TemporaryDirectory() as tmp: + Path(tmp, "icon.png").write_text("not an app icon") + app_dir = self._make_app(tmp, yaml, {"app.py": "pass"}) + valid, _config, _app_type, _warnings, errors = validate_app_dir(app_dir) + + self.assertFalse(valid) + self.assertTrue(any("icon file must stay within" in error.lower() for error in errors)) + + def test_source_directory_with_external_symlink_is_rejected(self): + yaml = """ +metadata: + name: Linked Source + bundle_id: org.test.linked-source + foreground: + process: + cmd: [python, app.py] +steps: + - name: Copy + type: files + files: + - source: ./files + destination: ./files +""" + with tempfile.TemporaryDirectory() as tmp: + app_dir = self._make_app(tmp, yaml, {"app.py": "pass"}) + source_dir = app_dir / "files" + source_dir.mkdir() + outside = Path(tmp, "secret.txt") + outside.write_text("secret", encoding="utf-8") + try: + (source_dir / "linked.txt").symlink_to(outside) + except OSError as exc: + self.skipTest(f"symlinks unavailable: {exc}") + + valid, _config, _app_type, _warnings, errors = validate_app_dir(app_dir) + + self.assertFalse(valid) + self.assertTrue(any("source file must stay within" in error.lower() for error in errors)) + def test_vnc_step_is_rejected_before_deploy(self): for step_type in ("vnc",): yaml = f""" diff --git a/tests/test_deploy_plan.py b/tests/test_deploy_plan.py index 084cf40..209b363 100644 --- a/tests/test_deploy_plan.py +++ b/tests/test_deploy_plan.py @@ -10,6 +10,7 @@ _bundle_id_from_name, _env_map_to_list, ) +from truffile.deploy.steps.files import handle_files class TestNormalizeCmd(unittest.TestCase): @@ -197,3 +198,41 @@ async def discard(self): self.assertTrue(client.discarded) self.assertIsNone(client.app_uuid) + + +class TestFileSteps(unittest.IsolatedAsyncioTestCase): + async def test_directory_upload_rejects_external_symlink(self): + class FakeClient: + async def upload(self, _src, _dest): + raise AssertionError("external symlink should not be uploaded") + + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + app_dir = base / "test-app" + source_dir = app_dir / "files" + source_dir.mkdir(parents=True) + outside = base / "outside.txt" + outside.write_text("secret", encoding="utf-8") + link = source_dir / "linked.txt" + try: + link.symlink_to(outside) + except OSError as exc: + self.skipTest(f"symlinks unavailable: {exc}") + + with self.assertRaisesRegex(ValueError, "Source file must stay within"): + await handle_files( + { + "files": [ + { + "source": "./files", + "destination": "./files", + } + ] + }, + client=FakeClient(), + app_dir=app_dir, + spinner_cls=_NoopSpinner, + arrow="->", + color_dim="", + color_reset="", + ) diff --git a/tests/test_obsidian_bridge.py b/tests/test_obsidian_bridge.py index c9d7066..533a693 100644 --- a/tests/test_obsidian_bridge.py +++ b/tests/test_obsidian_bridge.py @@ -55,3 +55,38 @@ def test_list_files_skips_hidden_entries(self): bridge = VaultBridge(root) entries = bridge.list_files("/") self.assertEqual(entries, ["visible/"]) + + def test_write_rejects_symlink_outside_vault(self): + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + root = base / "vault" + root.mkdir() + outside = base / "outside.md" + outside.write_text("outside", encoding="utf-8") + link = root / "linked.md" + try: + link.symlink_to(outside) + except OSError as exc: + self.skipTest(f"symlinks unavailable: {exc}") + + bridge = VaultBridge(root) + with self.assertRaises(ValueError): + bridge.write_note("linked.md", "overwrite") + + self.assertEqual(outside.read_text(encoding="utf-8"), "outside") + + def test_search_skips_symlink_outside_vault(self): + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + root = base / "vault" + root.mkdir() + outside = base / "outside.md" + outside.write_text("outside secret", encoding="utf-8") + link = root / "linked.md" + try: + link.symlink_to(outside) + except OSError as exc: + self.skipTest(f"symlinks unavailable: {exc}") + + bridge = VaultBridge(root) + self.assertEqual(bridge.search("outside secret"), []) diff --git a/truffile/cli/obsidian_bridge.py b/truffile/cli/obsidian_bridge.py index dce713b..1c3344b 100644 --- a/truffile/cli/obsidian_bridge.py +++ b/truffile/cli/obsidian_bridge.py @@ -11,6 +11,8 @@ from typing import Any from urllib.parse import parse_qs, urlparse +from truffile.path_safety import resolve_path_within + logger = logging.getLogger("truffile.obsidian_bridge") @@ -72,19 +74,12 @@ def normalize_relative_path(raw_path: str, *, allow_root: bool = False) -> Path: return Path(*parts) -def resolve_path_inside_vault(root: Path, raw_path: str, *, allow_root: bool = False, allow_missing: bool = False) -> Path: +def resolve_path_inside_vault(root: Path, raw_path: str, *, allow_root: bool = False) -> Path: rel = normalize_relative_path(raw_path, allow_root=allow_root) - candidate = root / rel - if allow_missing: - parent = candidate.parent.resolve() - if parent != root and root not in parent.parents: - raise ValueError("Path escapes the vault root") - return parent / candidate.name - - resolved = candidate.resolve() - if resolved != root and root not in resolved.parents: - raise ValueError("Path escapes the vault root") - return resolved + try: + return resolve_path_within(root, str(rel), label="Path") + except ValueError as exc: + raise ValueError("Path escapes the vault root") from exc class VaultBridge: @@ -129,7 +124,7 @@ def read_note(self, file_path: str) -> dict[str, Any]: } def write_note(self, file_path: str, content: str, *, append: bool = False) -> dict[str, Any]: - full = resolve_path_inside_vault(self.root, file_path, allow_missing=True) + full = resolve_path_inside_vault(self.root, file_path) full.parent.mkdir(parents=True, exist_ok=True) if append: with full.open("a", encoding="utf-8") as handle: @@ -157,8 +152,9 @@ def search(self, query: str, context_length: int = 100) -> list[dict[str, Any]]: if any(part.startswith(".") for part in rel.parts): continue try: - text = md_file.read_text(encoding="utf-8") - except Exception: + safe_file = resolve_path_inside_vault(self.root, rel.as_posix()) + text = safe_file.read_text(encoding="utf-8") + except (OSError, UnicodeError, ValueError): continue text_lower = text.lower() idx = text_lower.find(query_lower) diff --git a/truffile/deploy/steps/files.py b/truffile/deploy/steps/files.py index d50a726..322ea68 100644 --- a/truffile/deploy/steps/files.py +++ b/truffile/deploy/steps/files.py @@ -3,6 +3,7 @@ from pathlib import Path from typing import Any +from truffile.path_safety import resolve_path_within from truffile.transport.client import TruffleClient @@ -34,15 +35,20 @@ async def handle_files( **_kw: Any, ) -> None: for f in step.get("files", []): - src = app_dir / f["source"] + src = resolve_path_within(app_dir, f["source"], label="Source file") dest = f["destination"] if src.is_dir(): for child in sorted(src.rglob("*")): if child.is_file() and "__pycache__" not in str(child): + safe_child = resolve_path_within( + app_dir, + str(child.relative_to(app_dir)), + label="Source file", + ) rel = child.relative_to(src) child_dest = f"{dest.rstrip('/')}/{rel}" - await _upload_file(client, child, child_dest, spinner_cls, arrow, color_dim, color_reset) + await _upload_file(client, safe_child, child_dest, spinner_cls, arrow, color_dim, color_reset) elif src.is_file(): await _upload_file(client, src, dest, spinner_cls, arrow, color_dim, color_reset) else: diff --git a/truffile/path_safety.py b/truffile/path_safety.py new file mode 100644 index 0000000..e43ad4c --- /dev/null +++ b/truffile/path_safety.py @@ -0,0 +1,11 @@ +from __future__ import annotations + +from pathlib import Path + + +def resolve_path_within(root: Path, raw_path: str, *, label: str = "Path") -> Path: + resolved_root = root.resolve() + resolved = (resolved_root / raw_path).resolve(strict=False) + if resolved != resolved_root and resolved_root not in resolved.parents: + raise ValueError(f"{label} must stay within {resolved_root}") + return resolved diff --git a/truffile/schema/app_config.py b/truffile/schema/app_config.py index 8b5989f..7cbe545 100644 --- a/truffile/schema/app_config.py +++ b/truffile/schema/app_config.py @@ -7,6 +7,8 @@ import yaml +from truffile.path_safety import resolve_path_within + def _check_python_syntax(file_path: Path) -> tuple[bool, str]: try: @@ -147,9 +149,13 @@ def validate_app_dir(app_dir: Path) -> tuple[bool, dict[str, Any] | None, str | icon_file = meta.get("icon_file") if icon_file: - icon_path = app_dir / str(icon_file) - if not icon_path.exists(): - warnings.append(f"Icon file not found: {icon_file}") + try: + icon_path = resolve_path_within(app_dir, str(icon_file), label="Icon file") + except ValueError as exc: + errors.append(str(exc)) + else: + if not icon_path.exists(): + warnings.append(f"Icon file not found: {icon_file}") else: warnings.append("No icon specified in truffile.yaml") @@ -188,15 +194,30 @@ def validate_app_dir(app_dir: Path) -> tuple[bool, dict[str, Any] | None, str | for f in files_to_check: source = f.get("source") - if not isinstance(source, str): - errors.append("files entries must include a string 'source'") + if not isinstance(source, str) or not source.strip(): + errors.append("files entries must include a non-empty string 'source'") continue - src = app_dir / source + try: + src = resolve_path_within(app_dir, source, label="Source file") + except ValueError as exc: + errors.append(str(exc)) + continue if not src.exists(): errors.append(f"Source file not found: {src}") continue + if src.is_dir(): + for child in src.rglob("*"): + try: + resolve_path_within( + app_dir, + str(child.relative_to(app_dir)), + label="Source file", + ) + except ValueError as exc: + errors.append(str(exc)) + if src.suffix == ".py": ok, err = _check_python_syntax(src) if not ok: