Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 72 additions & 0 deletions tests/test_app_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"""
Expand Down
39 changes: 39 additions & 0 deletions tests/test_deploy_plan.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
_bundle_id_from_name,
_env_map_to_list,
)
from truffile.deploy.steps.files import handle_files


class TestNormalizeCmd(unittest.TestCase):
Expand Down Expand Up @@ -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="",
)
35 changes: 35 additions & 0 deletions tests/test_obsidian_bridge.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"), [])
26 changes: 11 additions & 15 deletions truffile/cli/obsidian_bridge.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")

Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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)
Expand Down
10 changes: 8 additions & 2 deletions truffile/deploy/steps/files.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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:
Expand Down
11 changes: 11 additions & 0 deletions truffile/path_safety.py
Original file line number Diff line number Diff line change
@@ -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
33 changes: 27 additions & 6 deletions truffile/schema/app_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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")

Expand Down Expand Up @@ -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:
Expand Down