alex/headquarter
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: set SSH key ownership to container user with mode 600

Browse Source 
- Mount SSH keys as bind (not ro) so docker exec --user root can chown
- Add apply_ssh_permissions() to permission_fixer.py
- Call apply_ssh_permissions() after container start for all instance types
- Derive container user from home_dir (/root → root, /home/user → user)
- Tests: apply_ssh_permissions unit tests + start_instance integration tests

Quality gates: pytest 236 passed (6 pre-existing failures), tsc --noEmit clean
This commit is contained in:
Alex Blank
2026-05-29 14:04:52 +02:00
parent ceaed9af66
commit 03d22c4d06
4 changed files with 374 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/tests/unit/test_permission_fixer.py
+37
View File
@@ -7,6 +7,7 @@ import pytest
from src.services.permission_fixer import (
PermissionFixError,
apply_mount_permissions,
apply_ssh_permissions,
check_root_user_available,
_run_in_container,
)
@@ -132,6 +133,42 @@ class TestRunInContainer:
_run_in_container("abc123", ["chown", "x"], 10)
class TestApplySshPermissions:
"""Tests for apply_ssh_permissions."""
@patch("src.services.permission_fixer._run_in_container")
def test_applies_chown_chmod_and_file_mode(self, mock_run) -> None:
result = apply_ssh_permissions("abc123", "/home/user/.ssh", "user")
assert result["success"] is True
assert mock_run.call_count == 3
chown_call = mock_run.call_args_list[0]
chmod_call = mock_run.call_args_list[1]
file_mode_call = mock_run.call_args_list[2]
assert chown_call[0][1] == ["chown", "-R", "user:user", "/home/user/.ssh"]
assert chmod_call[0][1] == ["chmod", "700", "/home/user/.ssh"]
assert file_mode_call[0][1][0] == "sh"
assert "find /home/user/.ssh -name 'id_*' -type f -exec chmod 600" in file_mode_call[0][1][2]
@patch("src.services.permission_fixer._run_in_container")
def test_uses_root_user(self, mock_run) -> None:
result = apply_ssh_permissions("abc123", "/root/.ssh", "root")
assert result["success"] is True
chown_call = mock_run.call_args_list[0]
assert chown_call[0][1] == ["chown", "-R", "root:root", "/root/.ssh"]
@patch("src.services.permission_fixer._run_in_container")
def test_reports_failure(self, mock_run) -> None:
mock_run.side_effect = PermissionFixError("chown failed")
result = apply_ssh_permissions("abc123", "/home/user/.ssh", "user")
assert result["success"] is False
assert "chown failed" in result["error"]
class TestCheckRootUserAvailable:
"""Tests for check_root_user_available."""
apps/api/tests/unit/test_tool_instances_legacy.py
+245
View File
@@ -701,6 +701,251 @@ class TestStartInstanceLegacyFallback:
mock_execute_compose.assert_called_once()
class TestStartInstanceSshPermissions:
"""SSH key mounts trigger permission fixes after container starts."""
@patch("src.api.tool_instances.write_compose_file")
@patch("src.api.tool_instances.apply_ssh_permissions")
@patch("src.api.tool_instances.wait_for_container_running")
@patch("src.api.tool_instances.execute_compose_command")
@patch("src.api.tool_instances.get_container_id")
@patch("src.api.tool_instances.get_container_name")
@patch("src.api.tool_instances.connect_container_to_network")
@patch("src.api.tool_instances._sanitize_compose_file")
@patch("src.api.tool_instances._get_user")
@patch("src.api.tool_instances._get_owned_project")
async def test_manifest_instance_applies_ssh_permissions(
self,
mock_get_project,
mock_get_user,
mock_sanitize,
mock_connect_network,
mock_get_container_name,
mock_get_container_id,
mock_execute_compose,
mock_wait_container,
mock_apply_ssh,
mock_write_compose,
mock_session,
fake_user_id,
fake_project_id,
fake_repo_id,
fake_instance_id,
fake_tool_type_id,
) -> None:
"""Manifest instance with SSH keys calls apply_ssh_permissions."""
from src.models.tool_definition_manifest import ToolDefinitionManifest
manifest_id = uuid.uuid4()
ssh_key_id = str(uuid.uuid4())
mock_get_user.return_value = AsyncMock()
mock_get_project.return_value = AsyncMock()
mock_execute_compose.return_value = (0, "started", "")
mock_get_container_id.return_value = "abc123"
mock_get_container_name.return_value = "test-container"
mock_connect_network.return_value = True
mock_wait_container.return_value = {
"success": True,
"status": "running",
"waited_seconds": 0.5,
}
mock_apply_ssh.return_value = {"success": True, "error": None}
instance = ToolInstance(
id=fake_instance_id,
name="manifest-instance",
repository_id=fake_repo_id,
tool_type_id=fake_tool_type_id,
compose_path="/data/instances/manifest-instance/docker-compose.yml",
status="stopped",
clone_mode="mount",
ssh_key_ids=[ssh_key_id],
created_at=datetime.now(),
updated_at=datetime.now(),
)
tool_type = ToolType(
id=fake_tool_type_id,
name="manifest-tool",
display_name="Manifest Tool",
default_port=8080,
definition_type="manifest",
manifest_id=manifest_id,
dockerfile_template=None,
compose_template=None,
)
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
manifest_def = ToolDefinitionManifest(
id=manifest_id,
name="test-manifest",
display_name="Test Manifest",
interface_type="web",
manifest={"user": {"name": "user", "uid": 1001, "gid": 1001}},
)
ssh_key = SSHKey(
id=uuid.UUID(ssh_key_id),
user_id=fake_user_id,
name="test-key",
public_key="ssh-ed25519 AAA test@test",
private_key_encrypted="enc",
)
async def _get(model, pk):
if model is ToolInstance and pk == fake_instance_id:
return instance
if model is ToolType and pk == fake_tool_type_id:
return tool_type
if model is GitRepository and pk == fake_repo_id:
return repo
if model is User and pk == fake_user_id:
return User(id=fake_user_id, email="test@example.com")
if model is ToolDefinitionManifest and pk == manifest_id:
return manifest_def
if model is SSHKey and pk == uuid.UUID(ssh_key_id):
return ssh_key
return None
mock_session.get.side_effect = _get
with patch("os.path.exists", return_value=True):
with patch(
"src.api.tool_instances._prepare_manifest_instance"
) as mock_prepare:
mock_prepare.return_value = (
"headquarter/test:latest",
"services:\n app:\n image: test",
{"name": "test-manifest", "user": {"name": "user"}},
"/home/user",
)
result = await start_instance(
project_id=fake_project_id,
repo_id=fake_repo_id,
instance_id=fake_instance_id,
data=None,
user_id=fake_user_id,
session=mock_session,
)
assert result["status"] == "running"
mock_apply_ssh.assert_called_once_with("abc123", "/home/user/.ssh", "user")
@patch("src.api.tool_instances.apply_ssh_permissions")
@patch("src.api.tool_instances.wait_for_container_running")
@patch("src.api.tool_instances.execute_compose_command")
@patch("src.api.tool_instances.get_container_id")
@patch("src.api.tool_instances.get_container_name")
@patch("src.api.tool_instances.connect_container_to_network")
@patch("src.api.tool_instances._sanitize_compose_file")
@patch("src.api.tool_instances._get_user")
@patch("src.api.tool_instances._get_owned_project")
async def test_legacy_instance_applies_ssh_permissions(
self,
mock_get_project,
mock_get_user,
mock_sanitize,
mock_connect_network,
mock_get_container_name,
mock_get_container_id,
mock_execute_compose,
mock_wait_container,
mock_apply_ssh,
mock_session,
fake_user_id,
fake_project_id,
fake_repo_id,
fake_instance_id,
fake_tool_type_id,
) -> None:
"""Legacy instance with SSH keys calls apply_ssh_permissions."""
ssh_key_id = str(uuid.uuid4())
mock_get_user.return_value = AsyncMock()
mock_get_project.return_value = AsyncMock()
mock_execute_compose.return_value = (0, "started", "")
mock_get_container_id.return_value = "abc123"
mock_get_container_name.return_value = "test-container"
mock_connect_network.return_value = True
mock_wait_container.return_value = {
"success": True,
"status": "running",
"waited_seconds": 0.5,
}
mock_apply_ssh.return_value = {"success": True, "error": None}
instance = ToolInstance(
id=fake_instance_id,
name="legacy-instance",
repository_id=fake_repo_id,
tool_type_id=fake_tool_type_id,
compose_path="/data/instances/legacy-instance/docker-compose.yml",
status="stopped",
clone_mode="mount",
ssh_key_ids=[ssh_key_id],
created_at=datetime.now(),
updated_at=datetime.now(),
)
tool_type = ToolType(
id=fake_tool_type_id,
name="legacy-tool",
display_name="Legacy Tool",
default_port=8080,
definition_type="legacy",
manifest_id=None,
dockerfile_template=None,
compose_template="services:\n app:\n image: nginx",
)
repo = GitRepository(
id=fake_repo_id,
project_id=fake_project_id,
name="test-repo",
path="/data/repos/test-repo",
remote_url=None,
ssh_key_id=None,
)
ssh_key = SSHKey(
id=uuid.UUID(ssh_key_id),
user_id=fake_user_id,
name="test-key",
public_key="ssh-ed25519 AAA test@test",
private_key_encrypted="enc",
)
async def _get(model, pk):
if model is ToolInstance and pk == fake_instance_id:
return instance
if model is ToolType and pk == fake_tool_type_id:
return tool_type
if model is GitRepository and pk == fake_repo_id:
return repo
if model is User and pk == fake_user_id:
return User(id=fake_user_id, email="test@example.com")
if model is SSHKey and pk == uuid.UUID(ssh_key_id):
return ssh_key
return None
mock_session.get.side_effect = _get
with patch("os.path.exists", return_value=True):
result = await start_instance(
project_id=fake_project_id,
repo_id=fake_repo_id,
instance_id=fake_instance_id,
data=None,
user_id=fake_user_id,
session=mock_session,
)
assert result["status"] == "running"
mock_apply_ssh.assert_called_once_with("abc123", "/root/.ssh", "root")
class TestStartInstanceManifestBranch:
"""Manifest branch is taken ONLY when definition_type == 'manifest'."""