diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index f5018e18c0c9..92df07ed85f3 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -704,6 +704,7 @@ def _collect(): metadata=metadata, event_payload=getattr(ev, "payload", None), task=task, + board=board_slug, ) except Exception as art_exc: logger.debug( @@ -1078,6 +1079,7 @@ async def _deliver_kanban_artifacts( metadata: dict, event_payload: Optional[dict], task, + board: Optional[str] = None, ) -> None: """Upload artifact files referenced by a completed kanban task. @@ -1091,9 +1093,11 @@ async def _deliver_kanban_artifacts( 2. ``event_payload['summary']`` (truncated first line) 3. ``task.result`` (legacy fallback) - Files are deduplicated, missing files are silently skipped (the - path may have been mentioned for reference only), and delivery - errors are logged but do not break the notifier loop. + Files are restricted to the task's durable attachment directory, + deduplicated, and silently skipped when missing. Completion stages + workspace artifacts there before this notifier runs, so adapters never + receive a worker-mutable workspace path. Delivery errors are logged but + do not break the notifier loop. """ from pathlib import Path as _Path @@ -1141,6 +1145,47 @@ def _add(path: str) -> None: if not candidates: return + allowed_roots: list[_Path] = [] + task_id = getattr(task, "id", None) if task is not None else None + if task_id: + try: + from hermes_cli import kanban_db as _kb + + declared_root = _kb.task_attachments_dir( + str(task_id), board=board + ).expanduser() + lexical_root = _Path(os.path.abspath(os.fspath(declared_root))) + resolved_root = declared_root.resolve(strict=True) + # The durable root is an authority boundary, not just a + # containment hint. Refuse any symlinked component instead of + # resolving it into an attacker-selected external directory. + if lexical_root == resolved_root and resolved_root.is_dir(): + allowed_roots.append(resolved_root) + else: + logger.warning( + "kanban notifier: refusing unsafe task attachment root: %s", + declared_root, + ) + except (OSError, RuntimeError, ValueError): + pass + + confined: list[str] = [] + for candidate in candidates: + try: + resolved = _Path(candidate).resolve(strict=True) + if any(resolved.is_relative_to(root) for root in allowed_roots): + confined.append(str(resolved)) + else: + logger.warning( + "kanban notifier: skipping unstaged task artifact: %s", + candidate, + ) + except (OSError, RuntimeError): + continue + candidates = confined + if not candidates: + return + _IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".webp"} _VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".3gp"} diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index ce29dbe7e200..4e446fb79d6e 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -79,6 +79,7 @@ import secrets import shutil import sqlite3 +import stat import subprocess import sys import threading @@ -792,6 +793,24 @@ def task_attachments_dir(task_id: str, board: Optional[str] = None) -> Path: return attachments_root(board=board) / task_id +def _connection_board(conn: sqlite3.Connection) -> str: + """Derive board ownership from an open connection's main DB path.""" + try: + row = conn.execute("PRAGMA database_list").fetchone() + db_path = Path(row["file"]).resolve(strict=False) + default_db = (kanban_home() / "kanban.db").resolve(strict=False) + if db_path == default_db: + return DEFAULT_BOARD + relative = db_path.relative_to(boards_root().resolve(strict=False)) + if len(relative.parts) == 2 and relative.parts[1] == "kanban.db": + board = _normalize_board_slug(relative.parts[0]) + if board: + return board + except (OSError, RuntimeError, TypeError, ValueError): + pass + return get_current_board() + + def worker_logs_dir(board: Optional[str] = None) -> Path: """Return the directory under which per-task worker logs are written. @@ -5346,7 +5365,71 @@ def __init__(self, phantom: list[str], completing_task_id: str): class ArtifactPreservationError(RuntimeError): - """Raised when a declared scratch deliverable cannot be preserved.""" + """Raised when a declared completion artifact cannot be preserved.""" + + +def _scope_completion_artifacts( + conn: sqlite3.Connection, + task_id: str, + metadata: Optional[dict], +) -> Optional[dict]: + """Canonicalize declared artifacts below the task's workspace.""" + if not isinstance(metadata, dict) or "artifacts" not in metadata: + return metadata + raw_artifacts = metadata.get("artifacts") + if not isinstance(raw_artifacts, (list, tuple)): + raise ArtifactPreservationError("metadata.artifacts must be a list") + if not raw_artifacts: + return metadata + + row = conn.execute( + "SELECT workspace_path FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + if not row or not row["workspace_path"]: + raise ArtifactPreservationError( + "completion artifacts require an available task workspace" + ) + try: + workspace_root = Path(row["workspace_path"]).expanduser().resolve(strict=True) + except (OSError, RuntimeError) as exc: + raise ArtifactPreservationError( + "completion artifacts require an available task workspace" + ) from exc + if not workspace_root.is_dir(): + raise ArtifactPreservationError( + "completion artifacts require an available task workspace" + ) + + canonical: list[str] = [] + seen: set[str] = set() + for item in raw_artifacts: + if not isinstance(item, str) or not item.strip(): + raise ArtifactPreservationError( + "every completion artifact must be a non-empty path string" + ) + candidate = Path(item).expanduser() + if not candidate.is_absolute(): + candidate = workspace_root / candidate + try: + resolved = candidate.resolve(strict=True) + resolved.relative_to(workspace_root) + except (OSError, RuntimeError, ValueError) as exc: + raise ArtifactPreservationError( + f"declared artifact is outside its task workspace: {item}" + ) from exc + if not resolved.is_file(): + raise ArtifactPreservationError( + f"declared artifact is not a regular file: {item}" + ) + normalized = str(resolved) + if normalized not in seen: + seen.add(normalized) + canonical.append(normalized) + + updated = dict(metadata) + updated["artifacts"] = canonical + return updated def complete_task( @@ -5428,6 +5511,7 @@ def complete_task( metadata = _merge_completion_prose_artifacts( conn, task_id, metadata, summary=summary, result=result, ) + metadata = _scope_completion_artifacts(conn, task_id, metadata) with write_txn(conn): # Parent completion is a hard invariant even for direct human review # approval. A parent may have been reopened after this task entered @@ -5477,7 +5561,7 @@ def complete_task( if cur.rowcount != 1: return False if isinstance(metadata, dict): - _persist_scratch_completion_artifacts(conn, task_id, metadata) + _persist_completion_artifacts(conn, task_id, metadata) for stored_path in metadata.pop("_staged_artifacts", []): path = Path(stored_path) _insert_completion_attachment( @@ -5605,20 +5689,22 @@ def _merge_completion_prose_artifacts( summary: Optional[str], result: Optional[str], ) -> Optional[dict]: - """Promote existing scratch files named in legacy completion prose. + """Promote existing workspace files named in legacy completion prose. ``artifacts=[...]`` is preferred. Older workers only wrote an absolute - deliverable path in ``summary``/``result``; discover it while scratch still - exists so cleanup cannot erase the file the user was promised. + deliverable path in ``summary``/``result``; discover it before completion + so it can be staged outside the worker-controlled workspace. """ row = conn.execute( - "SELECT workspace_kind, workspace_path FROM tasks WHERE id = ?", + "SELECT workspace_path FROM tasks WHERE id = ?", (task_id,), ).fetchone() - if not row or row["workspace_kind"] != "scratch" or not row["workspace_path"]: + if not row or not row["workspace_path"]: return metadata workspace = Path(row["workspace_path"]).expanduser() - if not _is_managed_scratch_path(workspace): + try: + workspace_root = workspace.resolve(strict=True) + except (OSError, RuntimeError): return metadata text = "\n".join(part for part in (summary, result) if part) if not text: @@ -5628,8 +5714,13 @@ def _merge_completion_prose_artifacts( for match in re.finditer(prefix + r"(?:[/\\][^\s`\"'<>]+)", text): raw = match.group(0).rstrip(".,;:!?)]}") candidate = Path(raw) - if candidate.is_file(): - discovered.append(str(candidate)) + try: + resolved = candidate.resolve(strict=True) + resolved.relative_to(workspace_root) + except (OSError, RuntimeError, ValueError): + continue + if resolved.is_file(): + discovered.append(str(resolved)) if not discovered: return metadata updated = dict(metadata) if isinstance(metadata, dict) else {} @@ -5644,113 +5735,177 @@ def _merge_completion_prose_artifacts( return updated -def _persist_scratch_completion_artifacts( +def _persist_completion_artifacts( conn: sqlite3.Connection, task_id: str, metadata: dict, ) -> None: - """Copy scratch-workspace completion artifacts before cleanup removes them.""" + """Stage completion artifacts outside the worker-controlled workspace.""" raw_artifacts = metadata.get("artifacts") if not isinstance(raw_artifacts, (list, tuple)): return + if not raw_artifacts: + return row = conn.execute( - "SELECT workspace_kind, workspace_path FROM tasks WHERE id = ?", + "SELECT workspace_path FROM tasks WHERE id = ?", (task_id,), ).fetchone() - if not row or row["workspace_kind"] != "scratch" or not row["workspace_path"]: + if not row or not row["workspace_path"]: return workspace = Path(row["workspace_path"]).expanduser() - is_managed, board = _managed_scratch_path_info(workspace) - if not is_managed: - return + board = _connection_board(conn) + if ( + not task_id + or task_id in {".", ".."} + or "/" in task_id + or "\\" in task_id + or Path(task_id).name != task_id + ): + raise ArtifactPreservationError("task id cannot name an attachment directory") try: - workspace_root = workspace.resolve() - except OSError: - return + workspace_root = workspace.resolve(strict=True) + except (OSError, RuntimeError) as exc: + raise ArtifactPreservationError( + "completion artifacts require an available task workspace" + ) from exc attachment_dir = task_attachments_dir(task_id, board=board) persisted: list[str] = [] used_destinations: set[Path] = set() changed = False - def _discard_copies() -> None: - for copied in used_destinations: - try: - copied.unlink(missing_ok=True) - except OSError: - pass - try: - attachment_dir.rmdir() - except OSError: - pass - - for item in raw_artifacts: - artifact = str(item).strip() if isinstance(item, str) else "" - if not artifact: - continue - src = Path(artifact).expanduser() - try: - resolved_src = src.resolve() - except OSError: - persisted.append(artifact) - continue - - if not resolved_src.is_relative_to(workspace_root): - persisted.append(artifact) - continue - - if not src.is_file(): - _discard_copies() - raise ArtifactPreservationError( - f"declared scratch artifact is unavailable or not a regular file: {artifact}" - ) - - size = resolved_src.stat().st_size - if size > KANBAN_ATTACHMENT_MAX_BYTES: - _discard_copies() - raise ArtifactPreservationError( - f"declared scratch artifact exceeds the " - f"{KANBAN_ATTACHMENT_MAX_BYTES}-byte limit: {artifact}" - ) + try: + with _open_pinned_attachment_directory(attachment_dir) as ( + pinned_attachment_dir, + attachment_dir_fd, + ): + def _discard_copies() -> None: + for copied in used_destinations: + try: + if attachment_dir_fd is None: + copied.unlink(missing_ok=True) + else: + os.unlink(copied.name, dir_fd=attachment_dir_fd) + except OSError: + pass + + for item in raw_artifacts: + artifact = str(item).strip() if isinstance(item, str) else "" + if not artifact: + continue + src = Path(artifact).expanduser() + try: + resolved_src = src.resolve(strict=True) + except OSError as exc: + _discard_copies() + raise ArtifactPreservationError( + f"declared completion artifact is unavailable: {artifact}" + ) from exc + + if not resolved_src.is_relative_to(workspace_root): + _discard_copies() + raise ArtifactPreservationError( + f"declared artifact is outside its task workspace: {artifact}" + ) - dest: Optional[Path] = None - try: - attachment_dir.mkdir(parents=True, exist_ok=True) - dest = _unique_attachment_path(attachment_dir, resolved_src.name, used_destinations) - with resolved_src.open("rb") as source_file, dest.open("xb") as destination_file: - copied = 0 - while chunk := source_file.read(1024 * 1024): - copied += len(chunk) - if copied > KANBAN_ATTACHMENT_MAX_BYTES: - raise ArtifactPreservationError( - f"declared scratch artifact grew beyond the size limit: {artifact}" - ) - destination_file.write(chunk) - except Exception as exc: - if dest is not None: try: - dest.unlink(missing_ok=True) - except OSError: - pass - _discard_copies() - if isinstance(exc, ArtifactPreservationError): - raise - raise ArtifactPreservationError( - f"could not preserve declared scratch artifact {artifact}: {exc}" - ) from exc + expected_stat = os.stat(resolved_src, follow_symlinks=False) + except OSError as exc: + _discard_copies() + raise ArtifactPreservationError( + f"declared completion artifact is unavailable or not a regular file: {artifact}" + ) from exc + if not stat.S_ISREG(expected_stat.st_mode): + _discard_copies() + raise ArtifactPreservationError( + f"declared completion artifact is unavailable or not a regular file: {artifact}" + ) + if expected_stat.st_size > KANBAN_ATTACHMENT_MAX_BYTES: + _discard_copies() + raise ArtifactPreservationError( + f"declared completion artifact exceeds the " + f"{KANBAN_ATTACHMENT_MAX_BYTES}-byte limit: {artifact}" + ) - used_destinations.add(dest) - persisted.append(str(dest.resolve())) - changed = True + dest: Optional[Path] = None + destination_fd: Optional[int] = None + try: + dest, destination_fd = _open_unique_attachment_file( + pinned_attachment_dir, + attachment_dir_fd, + resolved_src.name, + used_destinations, + ) + source_flags = ( + os.O_RDONLY + | getattr(os, "O_BINARY", 0) + | getattr(os, "O_NOFOLLOW", 0) + ) + source_fd = os.open(resolved_src, source_flags) + with ( + os.fdopen(source_fd, "rb") as source_file, + os.fdopen(destination_fd, "wb") as destination_file, + ): + opened_stat = os.fstat(source_file.fileno()) + if not stat.S_ISREG(opened_stat.st_mode): + raise ArtifactPreservationError( + f"declared completion artifact is not a regular file: {artifact}" + ) + if ( + opened_stat.st_dev != expected_stat.st_dev + or opened_stat.st_ino != expected_stat.st_ino + ): + raise ArtifactPreservationError( + f"declared completion artifact changed before it could be copied: {artifact}" + ) + if opened_stat.st_size > KANBAN_ATTACHMENT_MAX_BYTES: + raise ArtifactPreservationError( + f"declared completion artifact exceeds the " + f"{KANBAN_ATTACHMENT_MAX_BYTES}-byte limit: {artifact}" + ) + copied = 0 + while chunk := source_file.read(1024 * 1024): + copied += len(chunk) + if copied > KANBAN_ATTACHMENT_MAX_BYTES: + raise ArtifactPreservationError( + f"declared completion artifact grew beyond the size limit: {artifact}" + ) + destination_file.write(chunk) + except Exception as exc: + if destination_fd is not None: + with contextlib.suppress(OSError): + os.close(destination_fd) + if dest is not None: + try: + if attachment_dir_fd is None: + dest.unlink(missing_ok=True) + else: + os.unlink(dest.name, dir_fd=attachment_dir_fd) + except OSError: + pass + _discard_copies() + if isinstance(exc, ArtifactPreservationError): + raise + raise ArtifactPreservationError( + f"could not preserve declared completion artifact {artifact}: {exc}" + ) from exc + + used_destinations.add(dest) + persisted.append(str(dest)) + changed = True + except ArtifactPreservationError: + raise + except (OSError, RuntimeError, ValueError) as exc: + raise ArtifactPreservationError( + f"could not prepare the durable artifact directory: {exc}" + ) from exc if changed: metadata["artifacts"] = persisted - metadata["_staged_artifacts"] = [ - path for path in persisted if path.startswith(str(attachment_dir.resolve())) - ] + metadata["_staged_artifacts"] = list(persisted) def _insert_completion_attachment( @@ -5778,10 +5933,10 @@ def _insert_completion_attachment( def _unique_attachment_path(directory: Path, filename: str, used: set[Path]) -> Path: - """Return a non-conflicting path under ``directory`` for ``filename``.""" + """Return the next candidate path under ``directory`` for ``filename``.""" safe_name = Path(filename).name or "artifact" candidate = directory / safe_name - if candidate not in used and not candidate.exists(): + if candidate not in used: return candidate stem = Path(safe_name).stem or "artifact" @@ -5789,11 +5944,115 @@ def _unique_attachment_path(directory: Path, filename: str, used: set[Path]) -> idx = 1 while True: candidate = directory / f"{stem}_{idx}{suffix}" - if candidate not in used and not candidate.exists(): + if candidate not in used: return candidate idx += 1 +def _attachment_dir_fd_supported() -> bool: + """Return whether this platform supports descriptor-relative directory IO.""" + supported = getattr(os, "supports_dir_fd", set()) + return ( + hasattr(os, "O_DIRECTORY") + and hasattr(os, "O_NOFOLLOW") + and all(fn in supported for fn in (os.open, os.mkdir, os.unlink)) + ) + + +@contextlib.contextmanager +def _open_pinned_attachment_directory(directory: Path): + """Create and pin every destination component without following symlinks. + + On POSIX, every component is opened relative to its already-pinned parent + using ``O_NOFOLLOW``. The returned descriptor remains the authority for + file creation and cleanup even if the visible pathname is swapped later. + Platforms without descriptor-relative IO still reject every symlink via + ``lstat`` before opening the leaf; file creation remains ``O_EXCL`` and + ``O_NOFOLLOW`` where available. + """ + absolute = Path(os.path.abspath(os.fspath(directory.expanduser()))) + if not absolute.is_absolute() or not absolute.anchor: + raise ValueError("attachment directory must resolve to an absolute path") + + if not _attachment_dir_fd_supported(): + current = Path(absolute.anchor) + for component in absolute.parts[1:]: + current /= component + try: + os.mkdir(current, mode=0o700) + except FileExistsError: + pass + info = os.lstat(current) + if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode): + raise OSError(f"unsafe attachment directory component: {current}") + yield absolute, None + return + + directory_flags = ( + os.O_RDONLY + | os.O_DIRECTORY + | os.O_NOFOLLOW + | getattr(os, "O_CLOEXEC", 0) + ) + current_fd = os.open(absolute.anchor, directory_flags) + try: + for component in absolute.parts[1:]: + try: + os.mkdir(component, mode=0o700, dir_fd=current_fd) + except FileExistsError: + pass + next_fd = os.open(component, directory_flags, dir_fd=current_fd) + try: + if not stat.S_ISDIR(os.fstat(next_fd).st_mode): + raise NotADirectoryError(component) + except Exception: + os.close(next_fd) + raise + previous_fd = current_fd + current_fd = next_fd + os.close(previous_fd) + yield absolute, current_fd + finally: + os.close(current_fd) + + +def _open_unique_attachment_file( + directory: Path, + directory_fd: Optional[int], + filename: str, + used: set[Path], +) -> tuple[Path, int]: + """Atomically create a collision-free regular file in a pinned directory.""" + unavailable: set[Path] = set() + flags = ( + os.O_WRONLY + | os.O_CREAT + | os.O_EXCL + | getattr(os, "O_BINARY", 0) + | getattr(os, "O_NOFOLLOW", 0) + | getattr(os, "O_CLOEXEC", 0) + ) + while True: + destination = _unique_attachment_path( + directory, + filename, + used | unavailable, + ) + try: + if directory_fd is None: + fd = os.open(destination, flags, 0o600) + else: + fd = os.open(destination.name, flags, 0o600, dir_fd=directory_fd) + except FileExistsError: + unavailable.add(destination) + continue + opened = os.fstat(fd) + if not stat.S_ISREG(opened.st_mode): + os.close(fd) + raise OSError(f"attachment destination is not a regular file: {destination}") + return destination, fd + + def _managed_scratch_path_info(p: Path) -> tuple[bool, Optional[str]]: """Return whether *p* is managed scratch storage and the matching board.""" try: diff --git a/tests/gateway/test_kanban_watchers_mixin.py b/tests/gateway/test_kanban_watchers_mixin.py index 8454b5fd33dc..3ec93f5aa6fb 100644 --- a/tests/gateway/test_kanban_watchers_mixin.py +++ b/tests/gateway/test_kanban_watchers_mixin.py @@ -7,7 +7,12 @@ from __future__ import annotations +import asyncio import inspect +from pathlib import Path +from types import SimpleNamespace + +import pytest from gateway.kanban_watchers import GatewayKanbanWatchersMixin @@ -26,3 +31,160 @@ def test_mixin_defines_kanban_methods(): assert hasattr(GatewayKanbanWatchersMixin, m), f"mixin missing {m}" +class _ArtifactAdapter: + def __init__(self): + self.documents: list[str] = [] + + @staticmethod + def extract_local_files(content: str): + return ([part for part in content.split() if part.endswith(".pdf")], content) + + async def send_document(self, *, file_path: str, **_kwargs): + self.documents.append(file_path) + + +def test_notifier_rejects_unstaged_completion_paths(tmp_path): + workspace = tmp_path / "workspace" + workspace.mkdir() + inside = workspace / "inside.pdf" + outside = tmp_path / "outside.pdf" + inside.write_bytes(b"inside") + outside.write_bytes(b"outside") + adapter = _ArtifactAdapter() + task = SimpleNamespace( + id="t_bounded", + workspace_path=str(workspace), + result=f"legacy {outside}", + ) + + asyncio.run( + GatewayKanbanWatchersMixin()._deliver_kanban_artifacts( + adapter=adapter, + chat_id="chat", + metadata={}, + event_payload={ + "artifacts": [str(outside)], + "summary": f"deliver {inside} not {outside}", + }, + task=task, + ) + ) + + assert adapter.documents == [] + + +def test_durable_task_attachment_remains_deliverable( + tmp_path, + monkeypatch, +): + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + + from hermes_cli import kanban_db as kb + + task_id = "t_durable" + stored_dir = kb.task_attachments_dir(task_id, board="default") + stored_dir.mkdir(parents=True) + stored = stored_dir / "report.pdf" + stored.write_bytes(b"report") + adapter = _ArtifactAdapter() + task = SimpleNamespace( + id=task_id, + workspace_path=str(tmp_path / "cleaned-workspace"), + result=None, + ) + + asyncio.run( + GatewayKanbanWatchersMixin()._deliver_kanban_artifacts( + adapter=adapter, + chat_id="chat", + metadata={}, + event_payload={"artifacts": [str(stored)]}, + task=task, + board="default", + ) + ) + + assert adapter.documents == [str(stored.resolve())] + + +def test_notifier_rejects_symlinked_task_attachment_root( + tmp_path, + monkeypatch, +): + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + + from hermes_cli import kanban_db as kb + + task_id = "t_symlinked_root" + stored_dir = kb.task_attachments_dir(task_id, board="default") + stored_dir.parent.mkdir(parents=True) + outside = tmp_path / "outside" + outside.mkdir() + external = outside / "report.pdf" + external.write_bytes(b"host secret") + try: + stored_dir.symlink_to(outside, target_is_directory=True) + except OSError as exc: + pytest.skip(f"directory symlinks unavailable: {exc}") + + adapter = _ArtifactAdapter() + task = SimpleNamespace( + id=task_id, + workspace_path=str(tmp_path / "workspace"), + result=None, + ) + + asyncio.run( + GatewayKanbanWatchersMixin()._deliver_kanban_artifacts( + adapter=adapter, + chat_id="chat", + metadata={}, + event_payload={"artifacts": [str(stored_dir / external.name)]}, + task=task, + board="default", + ) + ) + + assert adapter.documents == [] + + +def test_notifier_never_hands_mutable_workspace_path_to_adapter(tmp_path): + workspace = tmp_path / "workspace" + workspace.mkdir() + artifact = workspace / "report.pdf" + outside = tmp_path / "outside.pdf" + artifact.write_bytes(b"expected report") + outside.write_bytes(b"host secret") + + class SwappingAdapter(_ArtifactAdapter): + def __init__(self): + super().__init__() + self.uploaded: list[bytes] = [] + + async def send_document(self, *, file_path: str, **_kwargs): + artifact.unlink() + artifact.symlink_to(outside) + self.uploaded.append(Path(file_path).read_bytes()) + + adapter = SwappingAdapter() + task = SimpleNamespace( + id="t_raced", + workspace_path=str(workspace), + result=None, + ) + + asyncio.run( + GatewayKanbanWatchersMixin()._deliver_kanban_artifacts( + adapter=adapter, + chat_id="chat", + metadata={}, + event_payload={"artifacts": [str(artifact)]}, + task=task, + ) + ) + + assert adapter.uploaded == [] diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index ce254bf9f5de..6e7bcec7d912 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -612,6 +612,264 @@ def test_complete_task_persists_scratch_artifacts_before_cleanup(kanban_home): ] +def test_complete_task_stages_dir_workspace_legacy_artifact( + kanban_home, + tmp_path, +): + """Persistent workspaces also stage a durable path outside worker control.""" + workspace = tmp_path / "project" + workspace.mkdir() + artifact = workspace / "report.pdf" + artifact.write_bytes(b"expected report") + + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="persistent report", + workspace_kind="dir", + workspace_path=str(workspace), + ) + assert kb.complete_task( + conn, + task_id, + summary=f"produced {artifact}", + ) + completed = [ + event for event in kb.list_events(conn, task_id) + if event.kind == "completed" + ][-1] + persisted = Path(completed.payload["artifacts"][0]) + run = kb.latest_run(conn, task_id) + attachments = kb.list_attachments(conn, task_id) + + assert artifact.exists(), "persistent workspace content must remain in place" + assert persisted.parent == kb.task_attachments_dir(task_id) + assert persisted != artifact + assert persisted.read_bytes() == b"expected report" + assert run is not None + assert run.metadata["artifacts"] == [str(persisted)] + assert [(item.filename, item.stored_path) for item in attachments] == [ + ("report.pdf", str(persisted.resolve())) + ] + + +def test_complete_task_stages_dir_artifact_on_connection_board( + kanban_home, + tmp_path, +): + """An explicit board connection owns staging even when current is default.""" + kb.create_board("other") + workspace = tmp_path / "other-project" + workspace.mkdir() + artifact = workspace / "report.pdf" + artifact.write_bytes(b"other board") + + with kb.connect(board="other") as conn: + task_id = kb.create_task( + conn, + title="other-board report", + workspace_kind="dir", + workspace_path=str(workspace), + ) + assert kb.complete_task( + conn, + task_id, + summary="done", + metadata={"artifacts": [str(artifact)]}, + ) + completed = [ + event for event in kb.list_events(conn, task_id) + if event.kind == "completed" + ][-1] + persisted = Path(completed.payload["artifacts"][0]) + + assert persisted.parent == kb.task_attachments_dir(task_id, board="other") + assert not kb.task_attachments_dir(task_id, board="default").exists() + + +def test_complete_task_does_not_clobber_existing_durable_artifact( + kanban_home, + tmp_path, +): + """Descriptor-relative collision handling must remain append-only.""" + workspace = tmp_path / "collision-project" + workspace.mkdir() + artifact = workspace / "report.pdf" + artifact.write_bytes(b"new report") + + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="collision-safe staging", + workspace_kind="dir", + workspace_path=str(workspace), + ) + destination = kb.task_attachments_dir(task_id) + destination.mkdir(parents=True) + existing = destination / artifact.name + existing.write_bytes(b"existing report") + + assert kb.complete_task( + conn, + task_id, + result="done", + metadata={"artifacts": [str(artifact)]}, + ) + completed = [ + event for event in kb.list_events(conn, task_id) + if event.kind == "completed" + ][-1] + persisted = Path(completed.payload["artifacts"][0]) + + assert existing.read_bytes() == b"existing report" + assert persisted.name == "report_1.pdf" + assert persisted.read_bytes() == b"new report" + + +@pytest.mark.parametrize("board", ["default", "other"]) +@pytest.mark.parametrize("force_path_fallback", [False, True]) +@pytest.mark.parametrize("symlink_level", ["attachments-root", "task-dir"]) +def test_complete_task_rejects_symlinked_destination_directory( + kanban_home, + tmp_path, + monkeypatch, + board, + force_path_fallback, + symlink_level, +): + """A pre-planted task attachment symlink must not redirect durable output.""" + if force_path_fallback: + monkeypatch.setattr(kb, "_attachment_dir_fd_supported", lambda: False) + if board != "default": + kb.create_board(board) + + workspace = tmp_path / f"{board}-project" + workspace.mkdir() + artifact = workspace / "report.pdf" + artifact.write_bytes(b"expected report") + + with kb.connect(board=board) as conn: + task_id = kb.create_task( + conn, + title=f"{board} destination confinement", + workspace_kind="dir", + workspace_path=str(workspace), + ) + destination = kb.task_attachments_dir(task_id, board=board) + outside = tmp_path / f"{board}-outside" + outside.mkdir() + if symlink_level == "attachments-root": + symlink = destination.parent + else: + destination.parent.mkdir(parents=True, exist_ok=True) + symlink = destination + symlink.parent.mkdir(parents=True, exist_ok=True) + try: + symlink.symlink_to(outside, target_is_directory=True) + except OSError as exc: + pytest.skip(f"directory symlinks unavailable: {exc}") + + with pytest.raises(kb.ArtifactPreservationError): + kb.complete_task( + conn, + task_id, + result="done", + metadata={"artifacts": [str(artifact)]}, + ) + + assert kb.get_task(conn, task_id).status == "ready" + assert kb.list_attachments(conn, task_id) == [] + assert all(event.kind != "completed" for event in kb.list_events(conn, task_id)) + + assert list(outside.iterdir()) == [] + assert artifact.read_bytes() == b"expected report" + + +@pytest.mark.parametrize("via_workspace_symlink", [False, True]) +def test_complete_task_rejects_artifacts_outside_scratch_workspace( + kanban_home, + tmp_path, + via_workspace_symlink, +): + """A completion artifact cannot turn the notifier into a host-file reader.""" + outside = tmp_path / "outside-secret.txt" + outside.write_text("host secret\n", encoding="utf-8") + + with kb.connect() as conn: + task_id = kb.create_task(conn, title="bounded artifact") + task = kb.get_task(conn, task_id) + workspace = kb.resolve_workspace(task) + kb.set_workspace_path(conn, task_id, workspace) + declared = outside + if via_workspace_symlink: + declared = workspace / "artifact.txt" + declared.symlink_to(outside) + + with pytest.raises( + kb.ArtifactPreservationError, + match="outside its task workspace", + ): + kb.complete_task( + conn, + task_id, + result="done", + metadata={"artifacts": [str(declared)]}, + ) + + assert kb.get_task(conn, task_id).status == "ready" + assert kb.list_attachments(conn, task_id) == [] + assert all(event.kind != "completed" for event in kb.list_events(conn, task_id)) + + assert workspace.exists(), "rejected completion must preserve the workspace" + assert outside.read_text(encoding="utf-8") == "host secret\n" + + +@pytest.mark.skipif( + not hasattr(os, "O_NOFOLLOW"), + reason="platform has no O_NOFOLLOW support", +) +def test_complete_task_rejects_artifact_swapped_to_symlink_before_open( + kanban_home, + tmp_path, + monkeypatch, +): + """Opening the source must not follow a symlink installed after validation.""" + outside = tmp_path / "outside-secret.txt" + outside.write_text("host secret\n", encoding="utf-8") + + with kb.connect() as conn: + task_id = kb.create_task(conn, title="race-safe artifact") + task = kb.get_task(conn, task_id) + workspace = kb.resolve_workspace(task) + kb.set_workspace_path(conn, task_id, workspace) + artifact = workspace / "artifact.txt" + artifact.write_text("expected output\n", encoding="utf-8") + + original_unique_path = kb._unique_attachment_path + + def swap_before_open(directory, filename, used): + artifact.unlink() + artifact.symlink_to(outside) + return original_unique_path(directory, filename, used) + + monkeypatch.setattr(kb, "_unique_attachment_path", swap_before_open) + + with pytest.raises(kb.ArtifactPreservationError): + kb.complete_task( + conn, + task_id, + result="done", + metadata={"artifacts": [str(artifact)]}, + ) + + assert kb.get_task(conn, task_id).status == "ready" + assert kb.list_attachments(conn, task_id) == [] + assert all(event.kind != "completed" for event in kb.list_events(conn, task_id)) + + assert workspace.exists(), "rejected completion must preserve the workspace" + assert outside.read_text(encoding="utf-8") == "host secret\n" + + # --------------------------------------------------------------------------- diff --git a/tests/hermes_cli/test_kanban_notify.py b/tests/hermes_cli/test_kanban_notify.py index ec01f5a5d34d..8151132c8d62 100644 --- a/tests/hermes_cli/test_kanban_notify.py +++ b/tests/hermes_cli/test_kanban_notify.py @@ -1,4 +1,5 @@ import asyncio +import json import pytest from pathlib import Path @@ -810,12 +811,20 @@ async def test_notifier_artifact_delivery_skips_missing_files(kanban_home, tmp_p # companion test for the full explanation. monkeypatch.setenv("HERMES_MEDIA_ALLOW_DIRS", str(tmp_path)) - real_pdf = tmp_path / "real.pdf" + workspace = tmp_path / "workspace" + workspace.mkdir() + real_pdf = workspace / "real.pdf" real_pdf.write_bytes(b"%PDF-fake") conn = kb.connect() try: - tid = kb.create_task(conn, title="t", assignee="worker1") + tid = kb.create_task( + conn, + title="t", + assignee="worker1", + workspace_kind="dir", + workspace_path=str(workspace), + ) kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="chat1") finally: conn.close() @@ -825,11 +834,33 @@ async def test_notifier_artifact_delivery_skips_missing_files(kanban_home, tmp_p try: kt._handle_complete({ "summary": "one real, one ghost", - "artifacts": [str(real_pdf), "/tmp/definitely-does-not-exist.pdf"], + "artifacts": [str(real_pdf)], }) finally: os.environ.pop("HERMES_KANBAN_TASK", None) + # Simulate a stale completed event whose durable attachment disappeared + # before the notifier tick. New completions reject missing declarations; + # the notifier still skips a file that vanishes after staging. + conn = kb.connect() + try: + row = conn.execute( + "SELECT id, payload FROM task_events " + "WHERE task_id = ? AND kind = 'completed' " + "ORDER BY id DESC LIMIT 1", + (tid,), + ).fetchone() + payload = json.loads(row["payload"]) + missing = kb.task_attachments_dir(tid) / "vanished.pdf" + payload["artifacts"].append(str(missing)) + with kb.write_txn(conn): + conn.execute( + "UPDATE task_events SET payload = ? WHERE id = ?", + (json.dumps(payload), row["id"]), + ) + finally: + conn.close() + runner = object.__new__(GatewayRunner) runner._owns_kanban_dispatcher_lock = lambda: True runner._running = True diff --git a/tools/kanban_tools.py b/tools/kanban_tools.py index d49b53a2212b..6e9b7bb3a95b 100644 --- a/tools/kanban_tools.py +++ b/tools/kanban_tools.py @@ -1775,7 +1775,8 @@ def _board_schema_prop() -> dict[str, str]: "references are caught before they leak into downstream " "automation. If you produced deliverable files (charts, PDFs, " "spreadsheets, generated images), list their absolute paths " - "in ``artifacts`` — the gateway notifier will upload them as " + "inside this task's workspace in ``artifacts`` — the gateway " + "notifier will upload them as " "native attachments to the human who subscribed to the task, " "so the deliverable lands in their chat alongside the summary " "instead of being a path they have to fetch by hand." @@ -1834,20 +1835,21 @@ def _board_schema_prop() -> dict[str, str]: "items": {"type": "string"}, "description": ( "Optional list of absolute paths to deliverable " - "files you produced during this run — generated " + "files inside this task's workspace that you produced " + "during this run — generated " "charts, PDFs, spreadsheets, images, archives. " - "Examples: [\"/tmp/q3-revenue.png\", " - "\"/tmp/report.pdf\"]. The gateway notifier " + "Examples: [\"/workspace/q3-revenue.png\", " + "\"/workspace/report.pdf\"]. The gateway notifier " "uploads each path as a native attachment to the " "subscribed chat (images embed inline, everything " "else uploads as a file) so the deliverable " "lands with the completion notification. Skip " "intermediate scratch files and references that " - "are not the deliverable. The path must exist " - "on disk at completion. Files inside a managed scratch " - "workspace are copied to durable task attachments before " - "cleanup; a missing declared scratch artifact keeps the " - "task in-flight so you can fix the path and retry." + "are not the deliverable. Each path must exist and resolve " + "inside the task workspace at completion. Accepted files " + "are copied to durable task attachments before the " + "completion event is emitted; a missing declaration keeps " + "the task in-flight so you can fix the path and retry." ), }, "board": _board_schema_prop(),