433 lines
16 KiB
Python
433 lines
16 KiB
Python
"""
|
|
Tests for Secure Container Manager.
|
|
|
|
Following TDD approach - these tests are written before implementation.
|
|
Tests cover all requirements from todo.md section 1.3.2.
|
|
All tests use mocked Podman client (no actual containers needed).
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import MagicMock, Mock, patch, call
|
|
from pathlib import Path
|
|
|
|
|
|
def test_create_container_with_valid_params_succeeds():
|
|
"""Test that container creation with valid params succeeds."""
|
|
from mcp_forge.podman.containers import ContainerConfig, SecureContainerManager
|
|
from pod_executor.containers.client import PodmanClient
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.security.audit import AuditLogger, AuditEventType
|
|
from pod_executor.security.resource_limits import ResourceLimits
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
# Setup mocks
|
|
mock_podman_client = MagicMock(spec=PodmanClient)
|
|
mock_container = MagicMock()
|
|
mock_container.id = "abc123"
|
|
mock_podman_client.client.containers.create.return_value = mock_container
|
|
|
|
audit_logger = AuditLogger(Path("/tmp/test_audit.log"))
|
|
validator = OperationValidator(SecurityConfig(audit_log=Path("/tmp/test_audit.log")))
|
|
|
|
manager = SecureContainerManager(
|
|
podman_client=mock_podman_client,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
config = ContainerConfig(
|
|
image="mcp-forge/python:3.11",
|
|
command=["python", "-c", "print('hello')"],
|
|
resource_limits=ResourceLimits(memory="512m", storage="1g", cpu_quota=100000)
|
|
)
|
|
|
|
container_id = manager.create_container(config, session_id="test-session-1")
|
|
|
|
assert container_id == "abc123"
|
|
mock_podman_client.client.containers.create.assert_called_once()
|
|
|
|
|
|
def test_create_container_with_forbidden_params_raises_security_error():
|
|
"""Test that forbidden parameters raise SecurityError."""
|
|
from mcp_forge.podman.containers import ContainerConfig, SecureContainerManager
|
|
from pod_executor.containers.client import PodmanClient
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
mock_podman_client = MagicMock(spec=PodmanClient)
|
|
audit_logger = AuditLogger(Path("/tmp/test_audit.log"))
|
|
validator = OperationValidator(SecurityConfig(audit_log=Path("/tmp/test_audit.log")))
|
|
|
|
manager = SecureContainerManager(
|
|
podman_client=mock_podman_client,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
config = ContainerConfig(
|
|
image="mcp-forge/python:3.11",
|
|
command=["python", "-c", "print('hello')"]
|
|
)
|
|
|
|
# Try to override security params (should be caught in to_podman_params or validation)
|
|
with pytest.raises(SecurityError):
|
|
# This should fail validation
|
|
manager.create_container(config, session_id="test-session", privileged=True)
|
|
|
|
|
|
def test_create_container_with_invalid_image_raises_security_error():
|
|
"""Test that invalid/disallowed images raise SecurityError."""
|
|
from mcp_forge.podman.containers import ContainerConfig, SecureContainerManager
|
|
from pod_executor.containers.client import PodmanClient
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
mock_podman_client = MagicMock(spec=PodmanClient)
|
|
audit_logger = AuditLogger(Path("/tmp/test_audit.log"))
|
|
validator = OperationValidator(SecurityConfig(audit_log=Path("/tmp/test_audit.log")))
|
|
|
|
manager = SecureContainerManager(
|
|
podman_client=mock_podman_client,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
config = ContainerConfig(
|
|
image="evil/malicious:latest",
|
|
command=["python", "-c", "print('hello')"]
|
|
)
|
|
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
manager.create_container(config, session_id="test-session")
|
|
assert "image" in str(exc_info.value).lower()
|
|
|
|
|
|
def test_create_container_enforces_required_parameters():
|
|
"""Test that required security parameters are enforced."""
|
|
from mcp_forge.podman.containers import ContainerConfig
|
|
|
|
config = ContainerConfig(
|
|
image="mcp-forge/python:3.11",
|
|
command=["python", "-c", "print('hello')"]
|
|
)
|
|
|
|
params = config.to_podman_params()
|
|
|
|
# Check required security parameters
|
|
assert params["network_mode"] == "none"
|
|
assert params["read_only"] is True
|
|
assert "no-new-privileges" in params["security_opt"]
|
|
assert params["user"] == "1000:1000"
|
|
|
|
|
|
def test_resource_limits_are_applied_correctly():
|
|
"""Test that resource limits are correctly applied."""
|
|
from mcp_forge.podman.containers import ContainerConfig
|
|
from pod_executor.security.resource_limits import ResourceLimits
|
|
|
|
limits = ResourceLimits(
|
|
memory="1g",
|
|
storage="2g",
|
|
cpu_quota=200000
|
|
)
|
|
|
|
config = ContainerConfig(
|
|
image="mcp-forge/python:3.11",
|
|
resource_limits=limits
|
|
)
|
|
|
|
params = config.to_podman_params()
|
|
|
|
assert params["mem_limit"] == "1073741824" # 1GB in bytes as string
|
|
assert params["cpu_quota"] == 200000
|
|
assert params["storage_opt"]["size"] == "2147483648" # 2GB in bytes as string
|
|
|
|
|
|
def test_volume_mounts_are_validated():
|
|
"""Test that volume mounts are validated against allowlist."""
|
|
from mcp_forge.podman.containers import ContainerConfig, SecureContainerManager
|
|
from pod_executor.containers.client import PodmanClient
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
mock_podman_client = MagicMock(spec=PodmanClient)
|
|
audit_logger = AuditLogger(Path("/tmp/test_audit.log"))
|
|
validator = OperationValidator(SecurityConfig(audit_log=Path("/tmp/test_audit.log")))
|
|
|
|
manager = SecureContainerManager(
|
|
podman_client=mock_podman_client,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
# Valid mount (session path)
|
|
config = ContainerConfig(
|
|
image="mcp-forge/python:3.11",
|
|
volumes={"/mcp-forge/sessions/test-session-1/workspace": {"bind": "/workspace", "mode": "rw"}}
|
|
)
|
|
|
|
# This should succeed (valid session path)
|
|
mock_container = MagicMock()
|
|
mock_container.id = "abc123"
|
|
mock_podman_client.client.containers.create.return_value = mock_container
|
|
container_id = manager.create_container(config, session_id="test-session-1")
|
|
assert container_id == "abc123"
|
|
|
|
# Invalid mount (forbidden path)
|
|
config_bad = ContainerConfig(
|
|
image="mcp-forge/python:3.11",
|
|
volumes={"/etc/passwd": {"bind": "/tmp/passwd", "mode": "r"}}
|
|
)
|
|
|
|
with pytest.raises(SecurityError):
|
|
manager.create_container(config_bad, session_id="test-session-1")
|
|
|
|
|
|
def test_start_container_on_session_container_succeeds():
|
|
"""Test that starting a session container succeeds."""
|
|
from mcp_forge.podman.containers import SecureContainerManager
|
|
from pod_executor.containers.client import PodmanClient
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
mock_podman_client = MagicMock(spec=PodmanClient)
|
|
mock_container = MagicMock()
|
|
mock_podman_client.client.containers.get.return_value = mock_container
|
|
|
|
audit_logger = AuditLogger(Path("/tmp/test_audit.log"))
|
|
validator = OperationValidator(SecurityConfig(audit_log=Path("/tmp/test_audit.log")))
|
|
|
|
# Register container with validator
|
|
validator.register_session_container("abc123")
|
|
|
|
manager = SecureContainerManager(
|
|
podman_client=mock_podman_client,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
manager.start_container("abc123")
|
|
mock_container.start.assert_called_once()
|
|
|
|
|
|
def test_start_container_on_non_session_container_raises_security_error():
|
|
"""Test that starting a non-session container raises SecurityError."""
|
|
from mcp_forge.podman.containers import SecureContainerManager
|
|
from pod_executor.containers.client import PodmanClient
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
mock_podman_client = MagicMock(spec=PodmanClient)
|
|
audit_logger = AuditLogger(Path("/tmp/test_audit.log"))
|
|
validator = OperationValidator(SecurityConfig(audit_log=Path("/tmp/test_audit.log")))
|
|
|
|
manager = SecureContainerManager(
|
|
podman_client=mock_podman_client,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
# Try to start container not registered with validator
|
|
with pytest.raises(SecurityError):
|
|
manager.start_container("unknown123")
|
|
|
|
|
|
def test_stop_container_works():
|
|
"""Test that stopping a container works."""
|
|
from mcp_forge.podman.containers import SecureContainerManager
|
|
from pod_executor.containers.client import PodmanClient
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
mock_podman_client = MagicMock(spec=PodmanClient)
|
|
mock_container = MagicMock()
|
|
mock_podman_client.client.containers.get.return_value = mock_container
|
|
|
|
audit_logger = AuditLogger(Path("/tmp/test_audit.log"))
|
|
validator = OperationValidator(SecurityConfig(audit_log=Path("/tmp/test_audit.log")))
|
|
|
|
# Register container
|
|
validator.register_session_container("abc123")
|
|
|
|
manager = SecureContainerManager(
|
|
podman_client=mock_podman_client,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
manager.stop_container("abc123", timeout=10)
|
|
mock_container.stop.assert_called_once_with(timeout=10)
|
|
|
|
|
|
def test_remove_container_works():
|
|
"""Test that removing a container works."""
|
|
from mcp_forge.podman.containers import SecureContainerManager
|
|
from pod_executor.containers.client import PodmanClient
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
mock_podman_client = MagicMock(spec=PodmanClient)
|
|
mock_container = MagicMock()
|
|
mock_podman_client.client.containers.get.return_value = mock_container
|
|
|
|
audit_logger = AuditLogger(Path("/tmp/test_audit.log"))
|
|
validator = OperationValidator(SecurityConfig(audit_log=Path("/tmp/test_audit.log")))
|
|
|
|
# Register container
|
|
validator.register_session_container("abc123")
|
|
|
|
manager = SecureContainerManager(
|
|
podman_client=mock_podman_client,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
manager.remove_container("abc123", force=True)
|
|
mock_container.remove.assert_called_once_with(force=True)
|
|
|
|
|
|
def test_cleanup_old_containers():
|
|
"""Test cleanup of old containers."""
|
|
from mcp_forge.podman.containers import SecureContainerManager
|
|
from pod_executor.containers.client import PodmanClient
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
mock_podman_client = MagicMock(spec=PodmanClient)
|
|
|
|
# Create mock old and new containers
|
|
old_container = MagicMock()
|
|
old_container.id = "old123"
|
|
old_container.attrs = {
|
|
"Created": (datetime.now() - timedelta(hours=25)).isoformat(),
|
|
"Labels": {"mcp-forge.session": "old-session"}
|
|
}
|
|
|
|
new_container = MagicMock()
|
|
new_container.id = "new123"
|
|
new_container.attrs = {
|
|
"Created": datetime.now().isoformat(),
|
|
"Labels": {"mcp-forge.session": "new-session"}
|
|
}
|
|
|
|
mock_podman_client.client.containers.list.return_value = [old_container, new_container]
|
|
|
|
audit_logger = AuditLogger(Path("/tmp/test_audit.log"))
|
|
validator = OperationValidator(SecurityConfig(audit_log=Path("/tmp/test_audit.log")))
|
|
|
|
manager = SecureContainerManager(
|
|
podman_client=mock_podman_client,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
count = manager.cleanup_old_containers(max_age=timedelta(hours=24))
|
|
|
|
assert count == 1
|
|
old_container.remove.assert_called_once_with(force=True)
|
|
new_container.remove.assert_not_called()
|
|
|
|
|
|
def test_get_container_logs():
|
|
"""Test getting container logs."""
|
|
from mcp_forge.podman.containers import SecureContainerManager
|
|
from pod_executor.containers.client import PodmanClient
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
mock_podman_client = MagicMock(spec=PodmanClient)
|
|
mock_container = MagicMock()
|
|
mock_container.logs.return_value = b"stdout output\nstderr output"
|
|
mock_podman_client.client.containers.get.return_value = mock_container
|
|
|
|
audit_logger = AuditLogger(Path("/tmp/test_audit.log"))
|
|
validator = OperationValidator(SecurityConfig(audit_log=Path("/tmp/test_audit.log")))
|
|
|
|
# Register container
|
|
validator.register_session_container("abc123")
|
|
|
|
manager = SecureContainerManager(
|
|
podman_client=mock_podman_client,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
stdout, stderr = manager.get_container_logs("abc123", tail=100)
|
|
|
|
assert "output" in stdout or "output" in stderr
|
|
mock_container.logs.assert_called()
|
|
|
|
|
|
def test_wait_for_container():
|
|
"""Test waiting for container to exit."""
|
|
from mcp_forge.podman.containers import SecureContainerManager
|
|
from pod_executor.containers.client import PodmanClient
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.security.audit import AuditLogger
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
|
|
mock_podman_client = MagicMock(spec=PodmanClient)
|
|
mock_container = MagicMock()
|
|
mock_container.wait.return_value = {"StatusCode": 0}
|
|
mock_podman_client.client.containers.get.return_value = mock_container
|
|
|
|
audit_logger = AuditLogger(Path("/tmp/test_audit.log"))
|
|
validator = OperationValidator(SecurityConfig(audit_log=Path("/tmp/test_audit.log")))
|
|
|
|
# Register container
|
|
validator.register_session_container("abc123")
|
|
|
|
manager = SecureContainerManager(
|
|
podman_client=mock_podman_client,
|
|
validator=validator,
|
|
audit_logger=audit_logger
|
|
)
|
|
|
|
exit_code = manager.wait_for_container("abc123", timeout=300)
|
|
|
|
assert exit_code == 0
|
|
mock_container.wait.assert_called_once_with(timeout=300)
|
|
|
|
|
|
def test_container_config_to_podman_params_includes_all_security_settings():
|
|
"""Test that ContainerConfig.to_podman_params includes all required settings."""
|
|
from mcp_forge.podman.containers import ContainerConfig
|
|
from pod_executor.security.resource_limits import ResourceLimits
|
|
|
|
config = ContainerConfig(
|
|
image="mcp-forge/python:3.11",
|
|
command=["python", "-m", "test"],
|
|
environment={"VAR1": "value1"},
|
|
volumes={"/mcp-forge/sessions/sess-1/work": {"bind": "/workspace", "mode": "rw"}},
|
|
resource_limits=ResourceLimits(memory="512m", storage="1g", cpu_quota=100000),
|
|
working_dir="/workspace",
|
|
user="1000:1000"
|
|
)
|
|
|
|
params = config.to_podman_params()
|
|
|
|
# Required security settings
|
|
assert params["network_mode"] == "none"
|
|
assert params["read_only"] is True
|
|
assert "no-new-privileges" in params["security_opt"]
|
|
assert params["user"] == "1000:1000"
|
|
|
|
# Configuration passthrough
|
|
assert params["image"] == "mcp-forge/python:3.11"
|
|
assert params["command"] == ["python", "-m", "test"]
|
|
assert params["environment"] == {"VAR1": "value1"}
|
|
assert params["working_dir"] == "/workspace"
|
|
|
|
# Resource limits
|
|
assert "mem_limit" in params
|
|
assert "cpu_quota" in params
|