security(gateway): re-resolve skills_sync path constants per call

tests/test_profile_isolation_runtime.py

@@ -89,6 +89,54 @@ def test_lockfile_default_arg_resolves_active_profile(self, two_profiles):
         assert taps_b.path == prof_b / "skills" / ".hub" / "taps.json"
 
 
+class TestSkillsSyncPathResolution:
+    """tools/skills_sync.py path constants must reflect the active profile —
+    otherwise gateway/run.py's in-process sync_skills() call on every
+    profile's startup syncs bundled skills into whichever profile imported
+    this module first, under the multiplexed gateway."""
+
+    def test_skills_dir_follows_override(self, two_profiles):
+        prof_a, prof_b = two_profiles
+        import tools.skills_sync as ss
+
+        a_seen = _under_override(prof_a, lambda: ss._skills_dir())
+        b_seen = _under_override(prof_b, lambda: ss._skills_dir())
+
+        assert a_seen == prof_a / "skills"
+        assert b_seen == prof_b / "skills"
+        assert a_seen != b_seen
+
+    def test_manifest_file_follows_override(self, two_profiles):
+        prof_a, prof_b = two_profiles
+        import tools.skills_sync as ss
+
+        b_seen = _under_override(prof_b, lambda: ss._manifest_file())
+        assert b_seen == prof_b / "skills" / ".bundled_manifest"
+
+    def test_legacy_attribute_access_follows_override(self, two_profiles):
+        """External `from tools.skills_sync import SKILLS_DIR`-style access
+        (PEP 562 module __getattr__) must also reflect the active profile."""
+        prof_a, prof_b = two_profiles
+        import tools.skills_sync as ss
+
+        a_seen = _under_override(prof_a, lambda: Path(ss.SKILLS_DIR))
+        b_seen = _under_override(prof_b, lambda: Path(ss.HERMES_HOME))
+
+        assert a_seen == prof_a / "skills"
+        assert b_seen == prof_b
+
+    def test_monkeypatched_constant_still_wins(self, two_profiles, monkeypatch, tmp_path):
+        """The existing test seam (patch the module constant, see
+        tests/tools/test_skills_sync.py) is preserved."""
+        _prof_a, prof_b = two_profiles
+        import tools.skills_sync as ss
+
+        forced = tmp_path / "forced_skills"
+        monkeypatch.setattr("tools.skills_sync.SKILLS_DIR", forced)
+        seen = _under_override(prof_b, lambda: ss._skills_dir())
+        assert seen == forced
+
+
 class TestGatewayCacheDirResolution:
     """gateway/platforms/base.py cache getters must follow the active profile."""
 

tools/skills_sync.py

@@ -36,9 +36,55 @@
 logger = logging.getLogger(__name__)
 
 
-HERMES_HOME = get_hermes_home()
-SKILLS_DIR = HERMES_HOME / "skills"
-MANIFEST_FILE = SKILLS_DIR / ".bundled_manifest"
+# Resolved per-call (not frozen at import) so the profile override is
+# honored — an import-time constant pins every later sync_skills() call in
+# the process to whichever profile's HERMES_HOME was active when this module
+# was first imported, so a later-starting profile's gateway silently syncs
+# bundled skills into the FIRST profile's directory instead of its own under
+# the multiplexed gateway (single-process multi-profile runtimes, e.g. the
+# desktop tui_gateway). Legacy names (HERMES_HOME, SKILLS_DIR, MANIFEST_FILE)
+# are re-exposed via __getattr__ below (PEP 562) so external
+# `from tools.skills_sync import SKILLS_DIR` callers still work, and a test's
+# `patch("tools.skills_sync.SKILLS_DIR", ...)` sets a real module attribute
+# that shadows dynamic resolution entirely (mirrors tools/skills_hub.py's
+# already-fixed instance of this exact bug class).
+
+
+# _override lets a test-injected real module attribute (patch/monkeypatch on
+# SKILLS_DIR etc.) win over dynamic resolution; None means resolve live.
+def _override(name: str):
+    return globals().get(name)
+
+
+def _hermes_home() -> Path:
+    forced = _override("HERMES_HOME")
+    return Path(forced) if forced is not None else get_hermes_home()
+
+
+def _skills_dir() -> Path:
+    forced = _override("SKILLS_DIR")
+    return Path(forced) if forced is not None else _hermes_home() / "skills"
+
+
+def _manifest_file() -> Path:
+    forced = _override("MANIFEST_FILE")
+    return Path(forced) if forced is not None else _skills_dir() / ".bundled_manifest"
+
+
+_DYNAMIC_PATH_RESOLVERS = {
+    "HERMES_HOME": _hermes_home,
+    "SKILLS_DIR": _skills_dir,
+    "MANIFEST_FILE": _manifest_file,
+}
+
+
+def __getattr__(name: str):
+    """Resolve legacy path constants dynamically (PEP 562) so they reflect the
+    active profile override; a test's patch-set real attribute shadows it."""
+    resolver = _DYNAMIC_PATH_RESOLVERS.get(name)
+    if resolver is not None:
+        return resolver()
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
 
 # Marker file written by `hermes profile create --no-skills` (named profiles)
 # and by the installer's `--no-skills` flag (the default ~/.hermes profile).
@@ -101,11 +147,12 @@ def _read_manifest() -> Dict[str, str]:
     Handles both v1 (plain names) and v2 (name:hash) formats.
     v1 entries get an empty hash string which triggers migration on next sync.
     """
-    if not MANIFEST_FILE.exists():
+    manifest_file = _manifest_file()
+    if not manifest_file.exists():
         return {}
     try:
         result = {}
-        for line in MANIFEST_FILE.read_text(encoding="utf-8").splitlines():
+        for line in manifest_file.read_text(encoding="utf-8").splitlines():
             line = line.strip()
             if not line:
                 continue
@@ -133,7 +180,7 @@ def _read_suppressed_names() -> set:
 
         return read_suppressed_names()
     except Exception:
-        path = SKILLS_DIR / ".curator_suppressed"
+        path = _skills_dir() / ".curator_suppressed"
         if not path.exists():
             return set()
         names = set()
@@ -155,12 +202,13 @@ def _write_manifest(entries: Dict[str, str]):
     """
     import tempfile
 
-    MANIFEST_FILE.parent.mkdir(parents=True, exist_ok=True)
+    manifest_file = _manifest_file()
+    manifest_file.parent.mkdir(parents=True, exist_ok=True)
     data = "\n".join(f"{name}:{hash_val}" for name, hash_val in sorted(entries.items())) + "\n"
 
     try:
         fd, tmp_path = tempfile.mkstemp(
-            dir=str(MANIFEST_FILE.parent),
+            dir=str(manifest_file.parent),
             prefix=".bundled_manifest_",
             suffix=".tmp",
         )
@@ -169,15 +217,15 @@ def _write_manifest(entries: Dict[str, str]):
                 f.write(data)
                 f.flush()
                 os.fsync(f.fileno())
-            atomic_replace(tmp_path, MANIFEST_FILE)
+            atomic_replace(tmp_path, manifest_file)
         except BaseException:
             try:
                 os.unlink(tmp_path)
             except OSError:
                 pass
             raise
     except Exception as e:
-        logger.debug("Failed to write skills manifest %s: %s", MANIFEST_FILE, e, exc_info=True)
+        logger.debug("Failed to write skills manifest %s: %s", manifest_file, e, exc_info=True)
 
 
 def _read_skill_name(skill_md: Path, fallback: str) -> str:
@@ -226,7 +274,7 @@ def _compute_relative_dest(skill_dir: Path, bundled_dir: Path) -> Path:
     e.g., bundled/skills/mlops/axolotl -> ~/.hermes/skills/mlops/axolotl
     """
     rel = skill_dir.relative_to(bundled_dir)
-    return SKILLS_DIR / rel
+    return _skills_dir() / rel
 
 
 def _dir_hash(directory: Path) -> str:
@@ -304,7 +352,7 @@ def _optional_skill_index() -> Dict[str, Tuple[str, str, Path]]:
 
 def _move_to_restore_backup(path: Path, backup_root: Path) -> str:
     """Move an existing skill directory into a restore backup, preserving rel path."""
-    rel = path.relative_to(SKILLS_DIR)
+    rel = path.relative_to(_skills_dir())
     target = backup_root / rel
     target.parent.mkdir(parents=True, exist_ok=True)
     if target.exists():
@@ -337,24 +385,25 @@ def restore_official_optional_skill(name: str, *, restore: bool = False) -> dict
     restored: List[str] = []
     backed_up: List[str] = []
     timestamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
-    backup_root = SKILLS_DIR / ".restore-backups" / f"official-optional-{timestamp}"
+    skills_dir = _skills_dir()
+    backup_root = skills_dir / ".restore-backups" / f"official-optional-{timestamp}"
 
     for folder_name, install_path, src in targets:
-        dest = SKILLS_DIR / Path(*install_path.split("/"))
+        dest = skills_dir / Path(*install_path.split("/"))
         src_hash = _dir_hash(src)
         canonical_ok = dest.exists() and _dir_hash(dest) == src_hash
 
         # Find already-active copies of this official skill by frontmatter name
         # or folder slug, even if curator moved it into another category.
         src_frontmatter = _read_skill_name(src / "SKILL.md", folder_name)
         matches: List[Path] = []
-        if SKILLS_DIR.exists():
-            for skill_md in sorted(SKILLS_DIR.rglob("SKILL.md")):
+        if skills_dir.exists():
+            for skill_md in sorted(skills_dir.rglob("SKILL.md")):
                 if is_excluded_skill_path(skill_md):
                     continue
                 candidate = skill_md.parent
                 try:
-                    candidate.relative_to(SKILLS_DIR)
+                    candidate.relative_to(skills_dir)
                 except ValueError:
                     continue
                 candidate_name = _read_skill_name(skill_md, candidate.name)
@@ -400,7 +449,7 @@ def _backfill_optional_provenance(quiet: bool = False) -> List[str]:
     if not optional_dir.exists():
         return []
 
-    lock_path = SKILLS_DIR / ".hub" / "lock.json"
+    lock_path = _skills_dir() / ".hub" / "lock.json"
     try:
         data = json.loads(lock_path.read_text()) if lock_path.exists() else {"version": 1, "installed": {}}
     except (json.JSONDecodeError, OSError):
@@ -423,7 +472,7 @@ def _backfill_optional_provenance(quiet: bool = False) -> List[str]:
         except ValueError as e:
             logger.debug("Skipping optional skill with unsafe path %s: %s", src, e)
             continue
-        dest = SKILLS_DIR / Path(*install_path.split("/"))
+        dest = _skills_dir() / Path(*install_path.split("/"))
         if not dest.exists() or not dest.is_dir():
             continue
         if _dir_hash(dest) != _dir_hash(src):
@@ -493,7 +542,7 @@ def sync_skills(quiet: bool = False) -> dict:
     # empty-result shape with skipped_opt_out lets callers report "opted out"
     # instead of "synced 0 / failed". This is the default-profile counterpart
     # to seed_profile_skills()'s marker check for named profiles.
-    if (HERMES_HOME / NO_BUNDLED_SKILLS_MARKER).exists():
+    if (_hermes_home() / NO_BUNDLED_SKILLS_MARKER).exists():
         if not quiet:
             print("  (skipped — profile opted out of bundled skills via .no-bundled-skills)")
         return {
@@ -510,7 +559,7 @@ def sync_skills(quiet: bool = False) -> dict:
             "optional_provenance_backfilled": [],
         }
 
-    SKILLS_DIR.mkdir(parents=True, exist_ok=True)
+    _skills_dir().mkdir(parents=True, exist_ok=True)
     manifest = _read_manifest()
     bundled_skills = _discover_bundled_skills(bundled_dir)
     bundled_names = {name for name, _ in bundled_skills}
@@ -696,7 +745,7 @@ def sync_skills(quiet: bool = False) -> dict:
     # Also copy DESCRIPTION.md files for categories (if not already present)
     for desc_md in bundled_dir.rglob("DESCRIPTION.md"):
         rel = desc_md.relative_to(bundled_dir)
-        dest_desc = SKILLS_DIR / rel
+        dest_desc = _skills_dir() / rel
         if not dest_desc.exists():
             try:
                 dest_desc.parent.mkdir(parents=True, exist_ok=True)
@@ -741,7 +790,7 @@ def _rmtree_writable(path: Path) -> None:
     # ``shutil.rmtree(~/.hermes)`` into a loud, recoverable ``ValueError``
     # instead of silently destroying the user's install.
     target = Path(path).resolve()
-    skills_root = SKILLS_DIR.resolve()
+    skills_root = _skills_dir().resolve()
     # Every legitimate caller passes a skill directory or its ``.bak``
     # sibling — always a strict child of the skills root. The skills root
     # itself must never be removed: a ``dest`` that collapses to
@@ -1049,11 +1098,11 @@ def set_bundled_skills_opt_out(enabled: bool) -> dict:
         dict with keys: ok (bool), changed (bool), marker (str path),
                         message (str).
     """
-    marker = HERMES_HOME / NO_BUNDLED_SKILLS_MARKER
+    marker = _hermes_home() / NO_BUNDLED_SKILLS_MARKER
     existed = marker.exists()
     try:
         if enabled:
-            HERMES_HOME.mkdir(parents=True, exist_ok=True)
+            _hermes_home().mkdir(parents=True, exist_ok=True)
             marker.write_text(
                 "This profile opted out of bundled-skill seeding "
                 "(`hermes skills opt-out`).\n"
@@ -1087,7 +1136,7 @@ def set_bundled_skills_opt_out(enabled: bool) -> dict:
 
 def is_bundled_skills_opt_out() -> bool:
     """Return True if the active profile carries the opt-out marker."""
-    return (HERMES_HOME / NO_BUNDLED_SKILLS_MARKER).exists()
+    return (_hermes_home() / NO_BUNDLED_SKILLS_MARKER).exists()
 
 
 def remove_pristine_bundled_skills(dry_run: bool = False) -> dict: