Skip to content
Merged
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
35 changes: 32 additions & 3 deletions src/vouch/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -352,6 +352,35 @@ def _deserialize_page(text: str) -> Page:
return Page(body=body, **meta)


def _validate_artifact_id(obj_id: str) -> str:
"""Reject artifact ids that would escape their subdirectory as filenames.

Ids are flat slugs (``claims/<id>.yaml``, ``pages/<id>.md``,
``sources/<id>/``, …). An untrusted proposer controls the id via
``slug_hint``, and the Claim/Page/Entity/Relation models do not
constrain ``id`` the way ``Source.id`` is hex-locked — so without this
guard, approving ``slug_hint="../../../../evil"`` writes an artifact
outside the KB, defeating the review gate.

Write-side counterpart to ``read_under_root`` and
``bundle._unsafe_name_reason``. Validating at the single point where
ids become path segments covers MCP, JSONL, CLI, and direct
``KBStore`` callers.
"""
if not obj_id or not isinstance(obj_id, str):
raise ValueError("artifact id must be a non-empty string")
if (
"/" in obj_id
or "\\" in obj_id
or "\x00" in obj_id
or os.path.isabs(obj_id)
or obj_id in (".", "..")
or ".." in Path(obj_id).parts
):
raise ValueError(f"unsafe artifact id (path traversal): {obj_id!r}")
return obj_id


def _load_page_or_skip(path: Path) -> Page | None:
"""Parse one page file; skip corrupt/unreadable files like ``_load_or_skip``."""
try:
Expand Down Expand Up @@ -549,16 +578,16 @@ def config_path(self) -> Path:
return self.kb_dir / CONFIG_FILENAME

def _yaml(self, sub: str, obj_id: str) -> Path:
return self.kb_dir / sub / f"{obj_id}.yaml"
return self.kb_dir / sub / f"{_validate_artifact_id(obj_id)}.yaml"

def _claim_path(self, claim_id: str) -> Path:
return self._yaml("claims", claim_id)

def _page_path(self, page_id: str) -> Path:
return self.kb_dir / "pages" / f"{page_id}.md"
return self.kb_dir / "pages" / f"{_validate_artifact_id(page_id)}.md"

def _source_dir(self, source_id: str) -> Path:
return self.kb_dir / "sources" / source_id
return self.kb_dir / "sources" / _validate_artifact_id(source_id)

def _entity_path(self, eid: str) -> Path:
return self._yaml("entities", eid)
Expand Down
51 changes: 51 additions & 0 deletions tests/test_storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -981,3 +981,54 @@ def test_list_pages_skips_unreadable_file(store: KBStore) -> None:

pages = store.list_pages()
assert [p.id for p in pages] == ["p-ok"]


# --- path-traversal: untrusted slug_hint / artifact id must not escape KB --


@pytest.mark.parametrize(
"bad_id",
["../evil", "..", "sub/evil", "a\\b", "/abs", ".", "x\x00y", ""],
)
def test_put_rejects_path_traversal_ids(store: KBStore, bad_id: str) -> None:
# Write builders must refuse ids that escape their subdirectory; an
# unsanitized id is a path-traversal write primitive.
src = store.put_source(b"e")
with pytest.raises(ValueError, match="artifact id"):
store.put_claim(Claim(id=bad_id, text="t", evidence=[src.id]))
with pytest.raises(ValueError, match="artifact id"):
store.put_page(Page(id=bad_id, title="T", body="b"))
with pytest.raises(ValueError, match="artifact id"):
store.put_entity(Entity(id=bad_id, name="N", type=EntityType.CONCEPT))
with pytest.raises(ValueError, match="artifact id"):
store.get_source(bad_id)


def test_validate_artifact_id_rejects_non_string() -> None:
from vouch.storage import _validate_artifact_id

with pytest.raises(ValueError, match="artifact id must be a non-empty string"):
_validate_artifact_id(None) # type: ignore[arg-type]


def test_approve_with_traversal_slug_hint_writes_nothing(
store: KBStore, tmp_path: Path
) -> None:
# End-to-end: an untrusted proposer supplies a malicious slug_hint; the
# proposal may file, but approval must not write an artifact outside the KB.
src = store.put_source(b"e")
slug_hint = "../../../../evil"
pr = propose_claim(
store,
text="t",
evidence=[src.id],
proposed_by="agent",
slug_hint=slug_hint,
)
# Exact target the unguarded join would have written.
escaped = (store.kb_dir / "claims" / f"{slug_hint}.yaml").resolve()
with pytest.raises((ProposalError, ValueError)):
approve(store, pr.id, approved_by="reviewer")
assert not escaped.exists()
assert not escaped.with_suffix("").exists()
assert list((store.kb_dir / "claims").glob("*.yaml")) == []
Loading