Restrict frontmatter updates to authorized keys

This commit is contained in:
酱牛肉 2026-09-09 19:32:09 +08:00
parent 5ac7d679fe
commit 4cd63d355d
8 changed files with 108 additions and 10 deletions

View file

@ -24,9 +24,13 @@ class BaseTagIndex(BaseComponent):
@tag_key.setter
def tag_key(self, value: object) -> None:
self._tag_key = self._validate_tag_key(value)
def _validate_tag_key(self, value: object) -> str:
"""Validate and normalize the configured frontmatter tag field."""
if not isinstance(value, str) or not value.strip():
raise ValueError("tag_key must be a non-empty string")
self._tag_key = value.strip()
return value.strip()
def set_healthy(self, healthy: bool) -> None:
"""Mark whether lookups can safely use the current derived state."""

View file

@ -19,11 +19,6 @@ class LocalTagIndex(BaseTagIndex):
max_tag_length: int = 64,
**kwargs,
):
if not isinstance(tag_key, str) or not tag_key.strip():
raise ValueError("tag_key must be a non-empty string")
tag_key = tag_key.strip()
if tag_key in FileFrontMatter.model_fields:
raise ValueError(f"tag_key must not be a reserved frontmatter key: {tag_key!r}")
super().__init__(tag_key=tag_key, **kwargs)
self.max_tags_per_file = self._positive_int("max_tags_per_file", max_tags_per_file)
self.max_tag_length = self._positive_int("max_tag_length", max_tag_length)
@ -31,6 +26,12 @@ class LocalTagIndex(BaseTagIndex):
self.tag_to_paths: dict[str, set[str]] = {}
self._maintenance_lock = asyncio.Lock()
def _validate_tag_key(self, value: object) -> str:
tag_key = super()._validate_tag_key(value)
if tag_key in FileFrontMatter.model_fields:
raise ValueError(f"tag_key must not be a reserved frontmatter key: {tag_key!r}")
return tag_key
@property
def n_files(self) -> int:
return len(self.path_to_tags)

View file

@ -552,7 +552,7 @@ jobs:
frontmatter_update:
backend: base
description: "Merge key-values into a file's frontmatter."
description: "Merge key-value pairs into a Markdown file's frontmatter while preserving its body and unrelated fields."
parameters:
type: object
properties:

View file

@ -78,7 +78,10 @@ class AutoTagStep(BaseStep):
self.prompt_format("user_message", path=path, tags_key=tag_key),
system_prompt=self.prompt_format("system_prompt", tags_key=tag_key),
job_tools=self.tools,
injected_job_kwargs={"_allowed_paths": [path]},
injected_job_kwargs={
"_allowed_paths": [path],
"_allowed_frontmatter_keys": [tag_key],
},
)
target = Path(self.file_store.workspace_path or ".") / path
@ -90,6 +93,7 @@ class AutoTagStep(BaseStep):
path=path,
metadata={tag_key: normalized},
_allowed_paths=[path],
_allowed_frontmatter_keys=[tag_key],
)
if not response.success:
raise RuntimeError(str(response.answer))

View file

@ -30,6 +30,10 @@ class FrontmatterUpdateStep(BaseStep):
injected by the server into the RuntimeContext;
without it, restricting read/edit/write alone would still leave
frontmatter of arbitrary workspace Markdown files mutable.
``_allowed_frontmatter_keys`` optionally limits updates to an injected
list of top-level keys. Omitting it (or setting it to ``None``) preserves
the unrestricted historical behavior.
"""
async def execute(self):
@ -39,6 +43,18 @@ class FrontmatterUpdateStep(BaseStep):
metadata = self.context.get("metadata") or {}
assert isinstance(metadata, dict), "metadata must be a dict"
allowed_keys = self.context.get("_allowed_frontmatter_keys")
key_error: str | None = None
if allowed_keys is not None:
if not isinstance(allowed_keys, list) or any(not isinstance(key, str) for key in allowed_keys):
key_error = "_allowed_frontmatter_keys must be a list of strings"
else:
allowed = set(allowed_keys)
denied = [key for key in metadata if not isinstance(key, str) or key not in allowed]
if denied:
names = ", ".join(sorted(repr(key) for key in denied))
key_error = f"frontmatter key(s) not allowed: {names}"
workspace_dir = Path(self.file_store.workspace_path or ".").resolve()
target, err = resolve_path(workspace_dir, path)
if err or target is None:
@ -57,6 +73,8 @@ class FrontmatterUpdateStep(BaseStep):
payload = {"path": path, "error": "not markdown"}
elif not metadata:
payload = {"path": path, "error": "no fields to update"}
elif key_error:
payload = {"path": path, "error": key_error}
else:
post = frontmatter.loads(target.read_text(encoding="utf-8"))
post.metadata.update(metadata)

View file

@ -95,8 +95,14 @@ async def test_auto_tag_filters_paths_and_continues_after_one_file_fails(tmp_pat
assert response.success is False
assert [call[1]["injected_job_kwargs"] for call in wrapper.calls] == [
{"_allowed_paths": ["daily/2026-09-09/failed.md"]},
{"_allowed_paths": ["daily/2026-09-09/first.md"]},
{
"_allowed_paths": ["daily/2026-09-09/failed.md"],
"_allowed_frontmatter_keys": ["tags"],
},
{
"_allowed_paths": ["daily/2026-09-09/first.md"],
"_allowed_frontmatter_keys": ["tags"],
},
]
assert all(
call[1]["job_tools"] == ["read", "list_tags", "frontmatter_read", "frontmatter_update"]
@ -131,6 +137,7 @@ async def test_auto_tag_uses_configured_key_and_normalizes_agent_output(tmp_path
async def update_frontmatter(name, /, **kwargs):
assert name == "frontmatter_update"
assert kwargs["_allowed_frontmatter_keys"] == ["keywords"]
post = frontmatter.loads(note.read_text(encoding="utf-8"))
post.metadata.update(kwargs["metadata"])
note.write_text(frontmatter.dumps(post), encoding="utf-8")

View file

@ -14,6 +14,7 @@ import os
import tempfile
from pathlib import Path
import frontmatter
import pytest
from reme.components.file_store import LocalFileStore
@ -89,6 +90,58 @@ async def test_update_no_suffix_autoappends_md():
await store.close()
@pytest.mark.asyncio
async def test_update_honors_injected_frontmatter_key_scope():
"""Injected key scope rejects the entire update when any key is outside it."""
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
note = _seed(Path(tmp), NOTE, BODY)
store = await _make_store()
before = note.read_bytes()
resp = await _run(
FrontmatterUpdateStep,
store,
path=NOTE,
metadata={"tags": ["new"], "name": "renamed"},
_allowed_frontmatter_keys=["tags"],
)
assert resp.success is False
assert resp.metadata["error"] == "frontmatter key(s) not allowed: 'name'"
assert note.read_bytes() == before
resp = await _run(
FrontmatterUpdateStep,
store,
path=NOTE,
metadata={"tags": ["new"]},
_allowed_frontmatter_keys=["tags"],
)
assert resp.success is True
assert frontmatter.loads(note.read_text(encoding="utf-8")).metadata == {"name": "n", "tags": ["new"]}
await store.close()
@pytest.mark.asyncio
async def test_update_empty_frontmatter_key_scope_denies_all_updates():
"""An empty injected list is restrictive, unlike an omitted constraint."""
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
note = _seed(Path(tmp), NOTE, BODY)
store = await _make_store()
before = note.read_bytes()
resp = await _run(
FrontmatterUpdateStep,
store,
path=NOTE,
metadata={"tags": ["new"]},
_allowed_frontmatter_keys=[],
)
assert resp.success is False
assert note.read_bytes() == before
await store.close()
@pytest.mark.asyncio
async def test_delete_no_suffix_autoappends_md():
"""frontmatter_delete on a suffix-less path resolves to the ``.md`` file."""

View file

@ -279,6 +279,17 @@ def test_tag_index_rejects_invalid_runtime_frontmatter_key() -> None:
assert index.tag_key == "tags"
@pytest.mark.parametrize("tag_key", ["name", "description"])
def test_tag_index_rejects_reserved_runtime_frontmatter_key(tag_key: str) -> None:
"""Runtime updates preserve the same reserved-field invariant as construction."""
index = LocalTagIndex()
with pytest.raises(ValueError, match="tag_key must not be a reserved frontmatter key"):
index.tag_key = tag_key
assert index.tag_key == "tags"
def test_file_store_updates_tag_index_from_file_nodes(monkeypatch, tmp_path: Path) -> None:
"""Keep daily and digest tags aligned through file-store mutations."""