fix(plugins): harden staged user-state carry

(cherry picked from commit 945d92c489227f0986884bcf35b0150f81e4674a)

hermes_cli/plugins_cmd_catalog.py

@@ -12,6 +12,7 @@ import json
 import logging
 import os
 import shutil
+import stat
 import sys
 import tempfile
 from pathlib import Path
@@ -22,6 +23,7 @@ from hermes_cli.plugin_catalog import (
     find_removed, get_live_catalog_entry, load_catalog_live, match_removed, resolved_removed_entries,
     _NAME_RE, _normalize_repo,
 )
+from pm.filesystem import is_junction
 
 logger = logging.getLogger(__name__)
 
@@ -364,46 +366,79 @@ def _carry_user_files(old: Path, new: Path, local: Optional[list[str]]) -> None:
         here = Path(dirpath)
         links = [name for name in dirnames if (here / name).is_symlink()]
         dirnames[:] = [
-            name for name in dirnames if name not in links and name not in _PRESERVE_SKIP
+            name
+            for name in dirnames
+            if name not in _PRESERVE_SKIP
+            and not (here / name).is_symlink()
+            and not is_junction(here / name)
         ]
         for name in (*filenames, *links):
             src = here / name
             rel = src.relative_to(old)
             if any(part in _PRESERVE_SKIP or part.endswith(".pyc") for part in rel.parts):
                 continue
+            try:
+                src_mode = src.lstat().st_mode
+            except OSError as exc:
+                raise PluginOperationError(f"Could not preserve user file '{rel}': {exc}") from exc
+            # FIFOs, sockets and devices are runtime objects, not durable plugin state.
+            if not (stat.S_ISREG(src_mode) or stat.S_ISLNK(src_mode)):
+                continue
+
             dst = new / rel
             if local is None:
                 # A no-git subdir install cannot distinguish removed upstream code from user files.
-                # Never resurrect old executable/control surfaces. This path may run from
-                # _install_plugin_core's post-scan before_swap hook, so do not inject an unscanned
-                # symlink either.
-                if src.is_symlink() or _revision_owned_without_git(rel):
+                # Never resurrect known executable/control surfaces, and never inject links after
+                # the installer's first scan.
+                if stat.S_ISLNK(src_mode) or _revision_owned_without_git(rel):
                     continue
                 if os.path.lexists(dst):
                     # A same-shape path belongs to the new revision when git cannot prove otherwise.
                     # A file -> directory clash is different: silently skipping it would delete a
                     # user-state file, so keep the live install intact and make the user resolve it.
+                    if is_junction(dst):
+                        raise PluginOperationError(
+                            f"Cannot preserve user file '{rel}': its destination conflicts with the "
+                            "updated plugin. The installed plugin was left unchanged."
+                        )
                     if dst.is_dir() and not dst.is_symlink():
                         raise _dir_clash(rel)
                     continue
             elif keep.isdisjoint((rel, *rel.parents)):
                 continue
 
+            if os.path.lexists(dst) and is_junction(dst):
+                raise PluginOperationError(
+                    f"Cannot preserve user file '{rel}': its destination conflicts with the "
+                    "updated plugin. The installed plugin was left unchanged."
+                )
             if dst.is_dir() and not dst.is_symlink():
                 raise _dir_clash(rel)
 
             parent = new
+            source_parent = old
             for part in rel.parent.parts:
                 parent /= part
+                source_parent /= part
                 if os.path.lexists(parent):
-                    if parent.is_symlink() or not parent.is_dir():
+                    if is_junction(parent) or parent.is_symlink() or not parent.is_dir():
                         raise PluginOperationError(
                             f"Cannot preserve user file '{rel}': its destination conflicts with the "
                             "updated plugin. The installed plugin was left unchanged."
                         )
                     continue
                 try:
-                    parent.mkdir()
+                    source_info = source_parent.lstat()
+                    if is_junction(source_parent) or not stat.S_ISDIR(source_info.st_mode):
+                        raise PluginOperationError(
+                            f"Cannot preserve user file '{rel}': its source path changed during the update. "
+                            "The installed plugin was left unchanged."
+                        )
+                    mode = stat.S_IMODE(source_info.st_mode)
+                    parent.mkdir(mode=mode)
+                    parent.chmod(mode)
+                except PluginOperationError:
+                    raise
                 except OSError as exc:
                     raise PluginOperationError(
                         f"Cannot preserve user file '{rel}': its destination could not be prepared. "

hermes_cli/plugins_cmd_install.py

@@ -339,6 +339,11 @@ def _install_plugin_core(
         _refuse_unavailable_portable_plugin(plugin_name, tmp_target)
         if before_swap is not None:
             before_swap(manifest, tmp_target)
+            # A callback may merge user-owned state into the candidate tree. Admit the final
+            # bytes through the same security and portable-package gates as the pristine clone.
+            _pc()._scan_plugin_tree(tmp_target, identifier, force=force, scan_decision_cb=scan_decision_cb,
+                                   reviewed_pin=at_reviewed_pin)
+            _refuse_unavailable_portable_plugin(plugin_name, tmp_target)
 
         if target.exists() and not force:
             raise _pc().PluginOperationError(

tests/hermes_cli/test_plugins_cmd_catalog.py

@@ -15,6 +15,7 @@ import pytest
 from hermes_cli import plugin_catalog as pc_cat
 from hermes_cli import plugins_cmd as pc
 from hermes_cli import plugins_cmd_catalog as cat
+from pm.filesystem import is_junction
 from tests.pm._fixtures import client, isolated_python  # noqa: F401
 
 pytestmark = pytest.mark.skipif(shutil.which("git") is None, reason="git not available")
@@ -282,6 +283,122 @@ def test_carry_user_files_fails_closed_on_staged_symlink_parent(tmp_path):
     assert (old / "data" / "index.db").read_text() == "user data"
 
 
+@pytest.mark.skipif(os.name == "nt", reason="POSIX directory modes")
+def test_carry_user_files_preserves_mode_of_created_directories(tmp_path):
+    """Directories created solely for carried state keep the source directory's restrictive mode."""
+    old, new = tmp_path / "old", tmp_path / "new"
+    (old / "data").mkdir(parents=True, mode=0o700)
+    (old / "data").chmod(0o700)
+    new.mkdir()
+    (old / "data" / "state.db").write_text("user data")
+
+    previous_umask = os.umask(0o022)
+    try:
+        cat._carry_user_files(old, new, None)
+    finally:
+        os.umask(previous_umask)
+
+    assert (new / "data" / "state.db").read_text() == "user data"
+    assert (new / "data").stat().st_mode & 0o777 == 0o700
+
+
+@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="FIFOs unavailable on this platform")
+def test_carry_user_files_skips_runtime_special_files(tmp_path):
+    """Runtime pipes are transient state and must not make an otherwise valid update fail."""
+    old, new = tmp_path / "old", tmp_path / "new"
+    (old / "data").mkdir(parents=True)
+    new.mkdir()
+    (old / "data" / "state.db").write_text("user data")
+    os.mkfifo(old / "data" / "events.fifo")
+
+    cat._carry_user_files(old, new, None)
+
+    assert (new / "data" / "state.db").read_text() == "user data"
+    assert not os.path.lexists(new / "data" / "events.fifo")
+
+
+@pytest.mark.skipif(os.name != "nt", reason="directory junctions are Windows-only")
+def test_carry_user_files_does_not_follow_source_junction(tmp_path):
+    """A source junction cannot pull files from outside the installed plugin into an update."""
+    old, new, outside = tmp_path / "old", tmp_path / "new", tmp_path / "outside"
+    old.mkdir()
+    new.mkdir()
+    outside.mkdir()
+    (outside / "secret.txt").write_text("outside")
+    junction = old / "data"
+    sp.run(["cmd", "/c", "mklink", "/J", str(junction), str(outside)], check=True, capture_output=True, text=True)
+    assert is_junction(junction)
+
+    cat._carry_user_files(old, new, None)
+
+    assert not os.path.lexists(new / "data")
+    assert not (new / "secret.txt").exists()
+
+
+@pytest.mark.skipif(os.name != "nt", reason="directory junctions are Windows-only")
+def test_carry_user_files_fails_closed_on_staged_junction_parent(tmp_path):
+    """A staged junction cannot redirect carried user data outside the replacement tree."""
+    old, new, outside = tmp_path / "old", tmp_path / "new", tmp_path / "outside"
+    (old / "data").mkdir(parents=True)
+    new.mkdir()
+    outside.mkdir()
+    (old / "data" / "index.db").write_text("user data")
+    junction = new / "data"
+    sp.run(["cmd", "/c", "mklink", "/J", str(junction), str(outside)], check=True, capture_output=True, text=True)
+    assert is_junction(junction)
+
+    with pytest.raises(pc.PluginOperationError, match="Cannot preserve user file"):
+        cat._carry_user_files(old, new, None)
+
+    assert not (outside / "index.db").exists()
+    assert (old / "data" / "index.db").read_text() == "user data"
+
+
+def test_url_subdir_reclone_revalidates_carried_code_before_publication(world, tmp_path, monkeypatch):
+    """The URL subdir path must run both admission gates again after carrying user state."""
+    from hermes_cli import plugins_cmd_install as install_cmd
+
+    mono = tmp_path / "mono-rescan"
+    src = mono / "plugins" / "sub-plugin"
+    src.mkdir(parents=True)
+    (src / "plugin.yaml").write_text("name: sub-plugin\nversion: 1.0.0\ndescription: d\n")
+    (src / "__init__.py").write_text("def register(ctx):\n    pass\n")
+    (src / "mcp.json").write_text(
+        '{"mcpServers":{"demo":{"type":"stdio","command":"${PLUGIN_ROOT}/server.js"}}}\n'
+    )
+    (src / "server.js").write_text('console.log("old revision")\n')
+    sp.run(["git", "init", "-q"], cwd=mono, check=True, env=_GIT_ENV)
+    _commit(mono, "v1")
+
+    target = pc._install_plugin_core(f"{mono.as_uri()}#plugins/sub-plugin", force=False)[0]
+    assert (target / "server.js").is_file()
+    (src / "server.js").unlink()
+    (src / "plugin.yaml").write_text("name: sub-plugin\nversion: 2.0.0\ndescription: d\n")
+    _commit(mono, "v2")
+
+    scans, portable_checks = [], []
+
+    def scan_gate(tree, *_args, **_kwargs):
+        scans.append((Path(tree) / "server.js").exists())
+
+    def portable_gate(_plugin_name, tree):
+        has_server = (Path(tree) / "server.js").exists()
+        portable_checks.append(has_server)
+        if has_server:
+            raise pc.PluginOperationError("carried server.js failed final admission")
+
+    monkeypatch.setattr(pc, "_scan_plugin_tree", scan_gate)
+    monkeypatch.setattr(install_cmd, "_refuse_unavailable_portable_plugin", portable_gate)
+    result = pc.dashboard_update_user_plugin("sub-plugin")
+
+    assert result["ok"] is False
+    assert "carried server.js failed final admission" in result["error"]
+    assert scans == [False, True]
+    assert portable_checks == [False, True]
+    assert (target / "server.js").is_file()
+    assert "version: 1.0.0" in (target / "plugin.yaml").read_text()
+
+
 def test_carry_user_files_fails_closed_when_source_tree_cannot_be_walked(tmp_path, monkeypatch):
     """Unreadable user state aborts replacement instead of being silently omitted."""
     old = tmp_path / "old"