486 lines
17 KiB
Python
486 lines
17 KiB
Python
"""
|
|
Tests for Podman operation allowlist and validation.
|
|
|
|
Following TDD approach - these tests are written before implementation.
|
|
Tests cover all security validation requirements from todo.md section 1.2.2.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
|
|
def test_allowed_operation_with_valid_params_passes():
|
|
"""Test that allowed operation with valid parameters passes validation."""
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(
|
|
audit_log=Path("/var/log/audit.log"),
|
|
enforce_resource_limits=True,
|
|
allow_network=False
|
|
)
|
|
validator = OperationValidator(config)
|
|
|
|
# Should not raise
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={
|
|
"network_mode": "none",
|
|
"read_only": True,
|
|
"security_opt": ["no-new-privileges"],
|
|
"user": "1000:1000",
|
|
"memory": "536870912",
|
|
"cpu_quota": 50000
|
|
},
|
|
session_id="test-session-123"
|
|
)
|
|
|
|
|
|
def test_allowed_operation_with_forbidden_params_raises_security_error():
|
|
"""Test that allowed operation with forbidden parameters raises SecurityError."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(
|
|
audit_log=Path("/var/log/audit.log"),
|
|
enforce_resource_limits=True,
|
|
allow_network=False
|
|
)
|
|
validator = OperationValidator(config)
|
|
|
|
# Try to add privileged mode
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={
|
|
"privileged": True, # FORBIDDEN
|
|
"network_mode": "none",
|
|
"read_only": True,
|
|
"security_opt": ["no-new-privileges"],
|
|
"user": "1000:1000"
|
|
},
|
|
session_id="test-session-123"
|
|
)
|
|
assert "privileged" in str(exc_info.value).lower()
|
|
|
|
|
|
def test_forbidden_param_cap_add_raises_security_error():
|
|
"""Test that cap_add parameter is rejected."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={
|
|
"cap_add": ["SYS_ADMIN"], # FORBIDDEN
|
|
"network_mode": "none",
|
|
"read_only": True
|
|
},
|
|
session_id="test-session"
|
|
)
|
|
assert "cap_add" in str(exc_info.value).lower()
|
|
|
|
|
|
def test_forbidden_param_devices_raises_security_error():
|
|
"""Test that devices parameter is rejected."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={
|
|
"devices": ["/dev/sda"], # FORBIDDEN
|
|
"network_mode": "none",
|
|
"read_only": True
|
|
},
|
|
session_id="test-session"
|
|
)
|
|
assert "devices" in str(exc_info.value).lower()
|
|
|
|
|
|
def test_required_parameters_validation():
|
|
"""Test that required parameters are enforced."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
# Missing network_mode
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={
|
|
"read_only": True,
|
|
"security_opt": ["no-new-privileges"],
|
|
"user": "1000:1000"
|
|
},
|
|
session_id="test-session"
|
|
)
|
|
assert "network_mode" in str(exc_info.value).lower()
|
|
|
|
# Missing read_only
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={
|
|
"network_mode": "none",
|
|
"security_opt": ["no-new-privileges"],
|
|
"user": "1000:1000"
|
|
},
|
|
session_id="test-session"
|
|
)
|
|
assert "read_only" in str(exc_info.value).lower()
|
|
|
|
|
|
def test_image_allowlist_enforcement():
|
|
"""Test that only allowed images can be used."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
# Valid image
|
|
validator.validate_image_name("mcp-forge/python:3.11")
|
|
validator.validate_image_name("mcp-forge/python:3.12")
|
|
validator.validate_image_name("mcp-forge/jupyter:latest")
|
|
validator.validate_image_name("mcp-forge/custom:my-env")
|
|
|
|
# Invalid image
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
validator.validate_image_name("ubuntu:latest")
|
|
assert "image" in str(exc_info.value).lower() or "allowlist" in str(exc_info.value).lower()
|
|
|
|
with pytest.raises(SecurityError):
|
|
validator.validate_image_name("malicious/image:latest")
|
|
|
|
|
|
def test_volume_mount_path_validation():
|
|
"""Test that volume mount paths are validated against allowlist."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
# Valid session path
|
|
validator.validate_volume_mount(
|
|
"/mcp-forge/sessions/test-session-123/workdir",
|
|
"test-session-123"
|
|
)
|
|
|
|
# Valid shared readonly path
|
|
validator.validate_volume_mount(
|
|
"/mcp-forge/shared/readonly/data",
|
|
"test-session-123"
|
|
)
|
|
|
|
# Valid upload path
|
|
validator.validate_volume_mount(
|
|
"/mcp-forge/uploads/test-session-123/file.txt",
|
|
"test-session-123"
|
|
)
|
|
|
|
# Invalid: root path
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
validator.validate_volume_mount("/", "test-session")
|
|
assert "forbidden" in str(exc_info.value).lower() or "root" in str(exc_info.value).lower()
|
|
|
|
# Invalid: /etc
|
|
with pytest.raises(SecurityError):
|
|
validator.validate_volume_mount("/etc/passwd", "test-session")
|
|
|
|
# Invalid: docker socket
|
|
with pytest.raises(SecurityError):
|
|
validator.validate_volume_mount("/var/run/docker.sock", "test-session")
|
|
|
|
# Invalid: podman socket
|
|
with pytest.raises(SecurityError):
|
|
validator.validate_volume_mount("/var/run/podman/podman.sock", "test-session")
|
|
|
|
|
|
def test_capability_restrictions():
|
|
"""Test that capability restrictions are enforced."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
# cap_add is forbidden
|
|
with pytest.raises(SecurityError):
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={
|
|
"cap_add": ["NET_ADMIN"],
|
|
"network_mode": "none",
|
|
"read_only": True
|
|
},
|
|
session_id="test-session"
|
|
)
|
|
|
|
|
|
def test_network_mode_enforcement():
|
|
"""Test that network mode is enforced as 'none'."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
# Wrong network mode
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={
|
|
"network_mode": "bridge", # Must be "none"
|
|
"read_only": True,
|
|
"security_opt": ["no-new-privileges"],
|
|
"user": "1000:1000"
|
|
},
|
|
session_id="test-session"
|
|
)
|
|
assert "network_mode" in str(exc_info.value).lower()
|
|
|
|
|
|
def test_privileged_mode_always_rejected():
|
|
"""Test that privileged mode is always rejected regardless of other params."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={
|
|
"privileged": True,
|
|
"network_mode": "none",
|
|
"read_only": True,
|
|
"security_opt": ["no-new-privileges"],
|
|
"user": "1000:1000"
|
|
},
|
|
session_id="test-session"
|
|
)
|
|
assert "privileged" in str(exc_info.value).lower()
|
|
|
|
|
|
def test_session_container_tracking():
|
|
"""Test that session containers are tracked and validated."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
# Register a session container
|
|
validator.register_session_container("container-123")
|
|
|
|
# Should be able to operate on registered container
|
|
validator.validate_container_start("container-123")
|
|
validator.validate_container_stop("container-123")
|
|
validator.validate_container_remove("container-123")
|
|
|
|
# Cannot operate on non-session container
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
validator.validate_container_start("unknown-container")
|
|
assert "session" in str(exc_info.value).lower() or "not found" in str(exc_info.value).lower()
|
|
|
|
with pytest.raises(SecurityError):
|
|
validator.validate_container_stop("unknown-container")
|
|
|
|
with pytest.raises(SecurityError):
|
|
validator.validate_container_remove("unknown-container")
|
|
|
|
|
|
def test_unregister_session_container():
|
|
"""Test that session containers can be unregistered."""
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
validator.register_session_container("container-123")
|
|
validator.validate_container_start("container-123") # Should work
|
|
|
|
validator.unregister_session_container("container-123")
|
|
|
|
# After unregistration, should not work
|
|
with pytest.raises(Exception): # SecurityError
|
|
validator.validate_container_start("container-123")
|
|
|
|
|
|
def test_validate_container_create_stores_container_id():
|
|
"""Test that validate_container_create automatically registers container."""
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
# Create container with session_id should auto-register
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={
|
|
"network_mode": "none",
|
|
"read_only": True,
|
|
"security_opt": ["no-new-privileges"],
|
|
"user": "1000:1000"
|
|
},
|
|
session_id="test-session-123"
|
|
)
|
|
|
|
# The actual container_id would be returned by Podman after creation
|
|
# So this test just verifies validation passes
|
|
|
|
|
|
def test_security_error_includes_rule_violation():
|
|
"""Test that SecurityError messages indicate what rule was violated."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
# Test various violations have clear messages
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={"privileged": True},
|
|
session_id="test"
|
|
)
|
|
error_msg = str(exc_info.value)
|
|
assert "privileged" in error_msg.lower()
|
|
|
|
with pytest.raises(SecurityError) as exc_info:
|
|
validator.validate_image_name("bad-image:latest")
|
|
error_msg = str(exc_info.value)
|
|
assert "image" in error_msg.lower() or "allowlist" in error_msg.lower()
|
|
|
|
|
|
def test_wildcard_image_pattern_matching():
|
|
"""Test that wildcard patterns work in image allowlist."""
|
|
from mcp_forge.security.allowlist import OperationValidator
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
# mcp-forge/custom:* should match any tag
|
|
validator.validate_image_name("mcp-forge/custom:my-env-v1")
|
|
validator.validate_image_name("mcp-forge/custom:another-tag")
|
|
validator.validate_image_name("mcp-forge/custom:abc123")
|
|
|
|
|
|
def test_forbidden_mount_paths_comprehensive():
|
|
"""Test all forbidden mount paths are blocked."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
forbidden_paths = [
|
|
"/",
|
|
"/etc",
|
|
"/etc/shadow",
|
|
"/var/run/docker.sock",
|
|
"/var/run/podman/podman.sock",
|
|
"/sys",
|
|
"/sys/kernel",
|
|
"/proc",
|
|
"/proc/self",
|
|
]
|
|
|
|
for path in forbidden_paths:
|
|
with pytest.raises(SecurityError):
|
|
validator.validate_volume_mount(path, "test-session")
|
|
|
|
|
|
def test_pid_mode_forbidden():
|
|
"""Test that pid_mode parameter is forbidden."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
with pytest.raises(SecurityError):
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={
|
|
"pid_mode": "host", # FORBIDDEN
|
|
"network_mode": "none"
|
|
},
|
|
session_id="test"
|
|
)
|
|
|
|
|
|
def test_ipc_mode_forbidden():
|
|
"""Test that ipc_mode parameter is forbidden."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
with pytest.raises(SecurityError):
|
|
validator.validate_container_create(
|
|
image="mcp-forge/python:3.11",
|
|
params={
|
|
"ipc_mode": "host", # FORBIDDEN
|
|
"network_mode": "none"
|
|
},
|
|
session_id="test"
|
|
)
|
|
|
|
|
|
def test_session_path_must_match_session_id():
|
|
"""Test that session paths must match the provided session_id."""
|
|
from mcp_forge.security.allowlist import OperationValidator, SecurityError
|
|
from mcp_forge.config.schema import SecurityConfig
|
|
from pathlib import Path
|
|
|
|
config = SecurityConfig(audit_log=Path("/audit.log"))
|
|
validator = OperationValidator(config)
|
|
|
|
# Correct session match
|
|
validator.validate_volume_mount(
|
|
"/mcp-forge/sessions/session-123/workdir",
|
|
"session-123"
|
|
)
|
|
|
|
# Wrong session in path
|
|
with pytest.raises(SecurityError):
|
|
validator.validate_volume_mount(
|
|
"/mcp-forge/sessions/other-session/workdir",
|
|
"session-123"
|
|
)
|