fix remote sync-back credential overwrite

tests/tools/test_file_sync.py

@@ -1,13 +1,15 @@
 """Tests for FileSyncManager — mtime tracking, deletion detection, transactional rollback."""
 
+import io
 import os
+import tarfile
 import time
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 
-from tools.environments.file_sync import FileSyncManager, _FORCE_SYNC_ENV
+from tools.environments.file_sync import FileSyncManager, _FORCE_SYNC_ENV, iter_sync_files
 
 
 @pytest.fixture
@@ -257,6 +259,60 @@ class TestEdgeCases:
         upload.assert_not_called()  # _file_mtime_key returns None, skipped
 
 
+class TestSyncBackSecurity:
+    def test_sync_back_does_not_overwrite_uploaded_credential_files(self, tmp_path, monkeypatch):
+        credential = tmp_path / "token.json"
+        credential.write_text("host-token", encoding="utf-8")
+        skill = tmp_path / "skill.py"
+        skill.write_text("host-skill", encoding="utf-8")
+
+        monkeypatch.setattr(
+            "tools.credential_files.get_credential_file_mounts",
+            lambda: [
+                {
+                    "host_path": str(credential),
+                    "container_path": "/root/.hermes/credentials/token.json",
+                }
+            ],
+        )
+        monkeypatch.setattr(
+            "tools.credential_files.iter_skills_files",
+            lambda container_base="/root/.hermes": [
+                {
+                    "host_path": str(skill),
+                    "container_path": f"{container_base}/skills/skill.py",
+                }
+            ],
+        )
+        monkeypatch.setattr(
+            "tools.credential_files.iter_cache_files",
+            lambda container_base="/root/.hermes": [],
+        )
+
+        def bulk_download(dest: Path) -> None:
+            with tarfile.open(dest, "w") as tar:
+                for name, data in {
+                    "root/.hermes/credentials/token.json": b"remote-token",
+                    "root/.hermes/skills/skill.py": b"remote-skill",
+                }.items():
+                    info = tarfile.TarInfo(name)
+                    info.size = len(data)
+                    tar.addfile(info, io.BytesIO(data))
+
+        mgr = FileSyncManager(
+            get_files_fn=lambda: iter_sync_files("/root/.hermes"),
+            upload_fn=MagicMock(),
+            delete_fn=MagicMock(),
+            bulk_download_fn=bulk_download,
+        )
+
+        mgr.sync(force=True)
+        mgr.sync_back(hermes_home=tmp_path)
+
+        assert credential.read_text(encoding="utf-8") == "host-token"
+        assert skill.read_text(encoding="utf-8") == "remote-skill"
+
+
 class TestBulkUpload:
     """Tests for the optional bulk_upload_fn callback."""
 

tools/environments/file_sync.py

@@ -76,6 +76,29 @@ def iter_sync_files(container_base: str = "/root/.hermes") -> list[tuple[str, st
     return files
 
 
+def _credential_host_paths() -> set[str]:
+    """Return credential files that are upload-only for remote sandboxes."""
+    try:
+        from tools.credential_files import get_credential_file_mounts
+    except Exception:
+        return set()
+
+    paths: set[str] = set()
+    try:
+        mounts = get_credential_file_mounts()
+    except Exception:
+        return set()
+    for entry in mounts:
+        host_path = entry.get("host_path") if isinstance(entry, dict) else None
+        if not host_path:
+            continue
+        try:
+            paths.add(str(Path(host_path).expanduser().resolve()))
+        except OSError:
+            paths.add(str(Path(host_path).expanduser()))
+    return paths
+
+
 def quoted_rm_command(remote_paths: list[str]) -> str:
     """Build a shell ``rm -f`` command for a batch of remote paths."""
     return "rm -f " + " ".join(shlex.quote(p) for p in remote_paths)
@@ -132,6 +155,7 @@ class FileSyncManager:
         self._delete_fn = delete_fn
         self._synced_files: dict[str, tuple[float, int]] = {}  # remote_path -> (mtime, size)
         self._pushed_hashes: dict[str, str] = {}  # remote_path -> sha256 hex digest
+        self._upload_only_host_paths: set[str] = set()
         self._last_sync_time: float = 0.0  # monotonic; 0 ensures first sync runs
         self._sync_interval = sync_interval
 
@@ -150,6 +174,7 @@ class FileSyncManager:
                 return
 
         current_files = self._get_files_fn()
+        self._upload_only_host_paths.update(_credential_host_paths())
         current_remote_paths = {remote for _, remote in current_files}
 
         # --- Uploads: new or changed files ---
@@ -328,6 +353,9 @@ class FileSyncManager:
                     tar.extractall(staging, filter="data")
 
                 applied = 0
+                upload_only_host_paths = (
+                    self._upload_only_host_paths | _credential_host_paths()
+                )
                 for dirpath, _dirnames, filenames in os.walk(staging):
                     for fname in filenames:
                         staged_file = os.path.join(dirpath, fname)
@@ -347,7 +375,11 @@ class FileSyncManager:
                         # Resolve host path from cached mapping
                         host_path = self._resolve_host_path(remote_path, file_mapping)
                         if host_path is None:
-                            host_path = self._infer_host_path(remote_path, file_mapping)
+                            host_path = self._infer_host_path(
+                                remote_path,
+                                file_mapping,
+                                upload_only_host_paths=upload_only_host_paths,
+                            )
                             if host_path is None:
                                 logger.debug(
                                     "sync_back: skipping %s (no host mapping)",
@@ -355,6 +387,13 @@ class FileSyncManager:
                                 )
                                 continue
 
+                        if self._is_upload_only_host_path(host_path, upload_only_host_paths):
+                            logger.debug(
+                                "sync_back: skipping upload-only credential file %s",
+                                remote_path,
+                            )
+                            continue
+
                         if os.path.exists(host_path) and pushed_hash is not None:
                             host_hash = _sha256_file(host_path)
                             if host_hash != pushed_hash:
@@ -384,7 +423,9 @@ class FileSyncManager:
         return None
 
     def _infer_host_path(self, remote_path: str,
-                         file_mapping: list[tuple[str, str]] | None = None) -> str | None:
+                         file_mapping: list[tuple[str, str]] | None = None,
+                         *,
+                         upload_only_host_paths: set[str] | None = None) -> str | None:
         """Infer a host path for a new remote file by matching path prefixes.
 
         Uses the existing file mapping to find a remote->host directory
@@ -394,10 +435,21 @@ class FileSyncManager:
         ``/root/.hermes/skills/b.md`` maps to ``~/.hermes/skills/b.md``.
         """
         mapping = file_mapping if file_mapping is not None else []
+        upload_only_host_paths = upload_only_host_paths or set()
         for host, remote in mapping:
+            if self._is_upload_only_host_path(host, upload_only_host_paths):
+                continue
             remote_dir = str(Path(remote).parent)
             if remote_path.startswith(remote_dir + "/"):
                 host_dir = str(Path(host).parent)
                 suffix = remote_path[len(remote_dir):]
                 return host_dir + suffix
         return None
+
+    @staticmethod
+    def _is_upload_only_host_path(host_path: str, upload_only_host_paths: set[str]) -> bool:
+        try:
+            resolved = str(Path(host_path).expanduser().resolve())
+        except OSError:
+            resolved = str(Path(host_path).expanduser())
+        return resolved in upload_only_host_paths