fix: prepare SSH keys with container UID/GID on host before mounting

- Extend prepare_ssh_key_files() with optional uid/gid parameters
- Call os.chown on created files when uid/gid are provided
- Gracefully handle PermissionError if API process is not root
- In start_instance, extract container user UID/GID from manifest
- Pass container UID/GID when preparing instance-level SSH key mounts
- Legacy clone-mode SSH keys continue to use root (0,0)
- Add unit tests for prepare_ssh_key_files ownership logic
- Keep apply_ssh_permissions() as fallback for cases where host chown fails

Quality gates: pytest 239 passed (6 pre-existing failures), tsc --noEmit clean

apps/api/src/api/tool_instances.py

@@ -1330,15 +1330,22 @@ async def start_instance(
     working_directory = None
     extra_volumes = []
 
-    # Fetch tool type early to determine home directory
+    # Fetch tool type early to determine home directory and container user
     tool_type = await session.get(ToolType, instance.tool_type_id)
     home_dir = "/root"
+    container_uid = 0
+    container_gid = 0
     if tool_type and tool_type.definition_type == "manifest" and tool_type.manifest_id:
         from src.models.tool_definition_manifest import ToolDefinitionManifest
 
         manifest_def = await session.get(ToolDefinitionManifest, tool_type.manifest_id)
         if manifest_def:
-            home_dir = get_manifest_home_dir(dict(manifest_def.manifest))
+            manifest = dict(manifest_def.manifest)
+            home_dir = get_manifest_home_dir(manifest)
+            user_cfg = manifest.get("user")
+            if user_cfg:
+                container_uid = user_cfg.get("uid", 0)
+                container_gid = user_cfg.get("gid", 0)
 
     # Apply selected config profile if any
     instance_dir = os.path.dirname(instance.compose_path)
@@ -1408,7 +1415,11 @@ async def start_instance(
             if ssh_key and ssh_key.user_id == user_id:
                 try:
                     ssh_dir = prepare_ssh_key_files(
-                        instance_dir, ssh_key, subdir=f"mounts/ssh/{key_id}/.ssh"
+                        instance_dir,
+                        ssh_key,
+                        subdir=f"mounts/ssh/{key_id}/.ssh",
+                        uid=container_uid,
+                        gid=container_gid,
                     )
                     ssh_target = os.path.join(home_dir, ".ssh")
                     extra_volumes.append(
@@ -1488,7 +1499,9 @@ async def start_instance(
                 ssh_key = await session.get(SSHKey, repo.ssh_key_id)
                 if ssh_key:
                     try:
-                        ssh_dir = prepare_ssh_key_files(instance_dir, ssh_key)
+                        ssh_dir = prepare_ssh_key_files(
+                            instance_dir, ssh_key, uid=0, gid=0
+                        )
                         extra_volumes.append(
                             {
                                 "source": ssh_dir,

apps/api/src/services/permission_fixer.py

@@ -208,7 +208,11 @@ def apply_ssh_permissions(
         )
         key_stat = _exec_and_log(
             container_id,
-            ["sh", "-c", f"stat -c '%U:%G %a %n' {ssh_target}/id_* 2>/dev/null || echo 'no id_* files found'"],
+            [
+                "sh",
+                "-c",
+                f"stat -c '%U:%G %a %n' {ssh_target}/id_* 2>/dev/null || echo 'no id_* files found'",
+            ],
             timeout,
             "verify-stat-keys",
         )

apps/api/src/services/ssh_keys.py

@@ -19,13 +19,21 @@ def _get_fernet() -> Fernet:
     return Fernet(key)
 
 
-def prepare_ssh_key_files(instance_dir: str, ssh_key, subdir: str = ".ssh") -> str:
+def prepare_ssh_key_files(
+    instance_dir: str,
+    ssh_key,
+    subdir: str = ".ssh",
+    uid: int | None = None,
+    gid: int | None = None,
+) -> str:
     """Decrypt and write SSH key files to instance directory for container mounting.
 
     Args:
         instance_dir: Path to instance directory
         ssh_key: SSHKey model instance with encrypted private key
         subdir: Subdirectory within instance_dir to write to (default: ".ssh")
+        uid: Optional UID to own the files (for bind-mount into non-root container)
+        gid: Optional GID to own the files
 
     Returns:
         Path to the .ssh directory
@@ -58,6 +66,20 @@ def prepare_ssh_key_files(instance_dir: str, ssh_key, subdir: str = ".ssh") -> s
     config_path.write_text(config_content)
     os.chmod(config_path, 0o644)
 
+    # Set ownership to target container user if requested
+    if uid is not None or gid is not None:
+        effective_uid = uid if uid is not None else -1
+        effective_gid = gid if gid is not None else -1
+        try:
+            os.chown(ssh_dir, effective_uid, effective_gid)
+            os.chown(private_key_path, effective_uid, effective_gid)
+            os.chown(public_key_path, effective_uid, effective_gid)
+            os.chown(config_path, effective_uid, effective_gid)
+        except PermissionError:
+            # API process may not be running as root; permission fixer will
+            # handle this post-start if the mount is read-write
+            pass
+
     return str(ssh_dir)
 
 

apps/api/tests/unit/test_permission_fixer.py

@@ -148,10 +148,32 @@ class TestApplySshPermissions:
         chmod_cmd = mock_run.call_args_list[1][0][0]
         file_mode_cmd = mock_run.call_args_list[2][0][0]
 
-        assert chown_cmd == ["docker", "exec", "--user", "root", "abc123", "chown", "-R", "user:user", "/home/user/.ssh"]
+        assert chown_cmd == [
-        assert chmod_cmd == ["docker", "exec", "--user", "root", "abc123", "chmod", "700", "/home/user/.ssh"]
+            "docker",
+            "exec",
+            "--user",
+            "root",
+            "abc123",
+            "chown",
+            "-R",
+            "user:user",
+            "/home/user/.ssh",
+        ]
+        assert chmod_cmd == [
+            "docker",
+            "exec",
+            "--user",
+            "root",
+            "abc123",
+            "chmod",
+            "700",
+            "/home/user/.ssh",
+        ]
         assert file_mode_cmd[0] == "docker"
-        assert "find /home/user/.ssh -name 'id_*' -type f -exec chmod 600" in file_mode_cmd[-1]
+        assert (
+            "find /home/user/.ssh -name 'id_*' -type f -exec chmod 600"
+            in file_mode_cmd[-1]
+        )
 
     @patch("subprocess.run")
     def test_uses_root_user(self, mock_run) -> None:
@@ -160,11 +182,23 @@ class TestApplySshPermissions:
 
         assert result["success"] is True
         chown_cmd = mock_run.call_args_list[0][0][0]
-        assert chown_cmd == ["docker", "exec", "--user", "root", "abc123", "chown", "-R", "root:root", "/root/.ssh"]
+        assert chown_cmd == [
+            "docker",
+            "exec",
+            "--user",
+            "root",
+            "abc123",
+            "chown",
+            "-R",
+            "root:root",
+            "/root/.ssh",
+        ]
 
     @patch("subprocess.run")
     def test_reports_failure(self, mock_run) -> None:
-        mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="chown failed")
+        mock_run.return_value = MagicMock(
+            returncode=1, stdout="", stderr="chown failed"
+        )
 
         result = apply_ssh_permissions("abc123", "/home/user/.ssh", "user")
 

apps/api/tests/unit/test_ssh_keys.py

@@ -0,0 +1,63 @@
+"""Unit tests for SSH key preparation."""
+
+import os
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from src.services.ssh_keys import prepare_ssh_key_files
+
+
+class TestPrepareSshKeyFiles:
+    """Tests for prepare_ssh_key_files."""
+
+    @patch("src.services.ssh_keys._get_fernet")
+    def test_creates_files_with_default_permissions(self, mock_fernet, tmp_path) -> None:
+        mock_fernet.return_value.decrypt.return_value = b"private-key-content"
+        ssh_key = MagicMock()
+        ssh_key.private_key_encrypted = "enc"
+        ssh_key.public_key = "ssh-ed25519 AAA test@test"
+
+        ssh_dir = prepare_ssh_key_files(str(tmp_path), ssh_key)
+
+        assert Path(ssh_dir).exists()
+        assert (Path(ssh_dir) / "id_ed25519").exists()
+        assert (Path(ssh_dir) / "id_ed25519.pub").exists()
+        assert (Path(ssh_dir) / "config").exists()
+        assert oct(os.stat(Path(ssh_dir) / "id_ed25519").st_mode)[-3:] == "600"
+
+    @patch("src.services.ssh_keys._get_fernet")
+    def test_sets_ownership_when_uid_gid_provided(self, mock_fernet, tmp_path) -> None:
+        mock_fernet.return_value.decrypt.return_value = b"private-key-content"
+        ssh_key = MagicMock()
+        ssh_key.private_key_encrypted = "enc"
+        ssh_key.public_key = "ssh-ed25519 AAA test@test"
+
+        with patch("os.chown") as mock_chown:
+            ssh_dir = prepare_ssh_key_files(
+                str(tmp_path), ssh_key, uid=1001, gid=1001
+            )
+
+        # os.chown is called for the directory and each of the 3 files
+        assert mock_chown.call_count == 4
+        # First call is the directory
+        assert mock_chown.call_args_list[0][0][1] == 1001
+        assert mock_chown.call_args_list[0][0][2] == 1001
+
+    @patch("src.services.ssh_keys._get_fernet")
+    def test_gracefully_handles_permission_error_on_chown(
+        self, mock_fernet, tmp_path
+    ) -> None:
+        mock_fernet.return_value.decrypt.return_value = b"private-key-content"
+        ssh_key = MagicMock()
+        ssh_key.private_key_encrypted = "enc"
+        ssh_key.public_key = "ssh-ed25519 AAA test@test"
+
+        with patch("os.chown", side_effect=PermissionError("not allowed")):
+            # Should not raise
+            ssh_dir = prepare_ssh_key_files(
+                str(tmp_path), ssh_key, uid=1001, gid=1001
+            )
+
+        assert Path(ssh_dir).exists()