HansAschauer-Privat/mcp-forge
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

initial commit after one day coding agent session

Browse source
This commit is contained in:
Hans Aschauer 2026-02-07 07:45:57 +01:00
commit 372af75b90
88 changed files with 22694 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

486
tests/security/test_allowlist.py Normal file
View file

@ -0,0 +1,486 @@
"""
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"
)

369
tests/security/test_audit.py Normal file
View file

@ -0,0 +1,369 @@
"""
Tests for audit logger module.
Following TDD approach - these tests are written before implementation.
Tests cover all logging requirements from todo.md section 1.2.3.
"""
import pytest
import json
from pathlib import Path
import tempfile
import threading
import time
from datetime import datetime
def test_log_entries_written_to_file(tmp_path):
"""Test that log entries are written to the log file."""
from mcp_forge.security.audit import AuditLogger, AuditEventType, AuditSeverity
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
logger.log(
event_type=AuditEventType.CONTAINER_CREATE,
severity=AuditSeverity.INFO,
message="Test container created",
details={"image": "test-image", "container_id": "abc123"}
)
assert log_file.exists()
content = log_file.read_text()
assert len(content) > 0
def test_log_entries_are_valid_json(tmp_path):
"""Test that log entries are valid JSON."""
from mcp_forge.security.audit import AuditLogger, AuditEventType, AuditSeverity
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
logger.log(
event_type=AuditEventType.CONTAINER_CREATE,
severity=AuditSeverity.INFO,
message="Test entry",
details={"key": "value"}
)
# Each line should be valid JSON
lines = log_file.read_text().strip().split("\n")
for line in lines:
data = json.loads(line) # Should not raise
assert isinstance(data, dict)
def test_log_entries_contain_required_fields(tmp_path):
"""Test that log entries contain all required fields."""
from mcp_forge.security.audit import AuditLogger, AuditEventType, AuditSeverity
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
logger.log(
event_type=AuditEventType.EXECUTION_REQUEST,
severity=AuditSeverity.INFO,
message="Code execution requested",
session_id="test-session-123"
)
lines = log_file.read_text().strip().split("\n")
entry = json.loads(lines[0])
# Required fields
assert "timestamp" in entry
assert "event_type" in entry
assert "severity" in entry
assert "message" in entry
assert "session_id" in entry
def test_timestamp_format_is_iso_8601(tmp_path):
"""Test that timestamp is in ISO 8601 format."""
from mcp_forge.security.audit import AuditLogger, AuditEventType, AuditSeverity
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
logger.log(
event_type=AuditEventType.SESSION_CREATE,
severity=AuditSeverity.INFO,
message="Session created"
)
lines = log_file.read_text().strip().split("\n")
entry = json.loads(lines[0])
# Should be parseable as ISO 8601
timestamp = entry["timestamp"]
dt = datetime.fromisoformat(timestamp)
assert isinstance(dt, datetime)
def test_concurrent_logging_is_thread_safe(tmp_path):
"""Test that concurrent logging from multiple threads is thread-safe."""
from mcp_forge.security.audit import AuditLogger, AuditEventType, AuditSeverity
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
def log_entries(thread_id, count):
for i in range(count):
logger.log(
event_type=AuditEventType.CONTAINER_CREATE,
severity=AuditSeverity.INFO,
message=f"Thread {thread_id} entry {i}"
)
# Create multiple threads
threads = []
entries_per_thread = 10
num_threads = 5
for i in range(num_threads):
t = threading.Thread(target=log_entries, args=(i, entries_per_thread))
threads.append(t)
t.start()
# Wait for all threads
for t in threads:
t.join()
# Verify all entries written
lines = log_file.read_text().strip().split("\n")
assert len(lines) == num_threads * entries_per_thread
# Verify all entries are valid JSON
for line in lines:
json.loads(line)
def test_security_violations_logged_with_correct_severity(tmp_path):
"""Test that security violations are logged at correct severity."""
from mcp_forge.security.audit import AuditLogger, AuditEventType, AuditSeverity
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
logger.log_security_violation(
operation="container_create",
reason="Privileged mode attempted",
session_id="test-session"
)
lines = log_file.read_text().strip().split("\n")
entry = json.loads(lines[0])
assert entry["severity"] == "critical"
assert entry["event_type"] == "security.violation"
def test_pii_is_not_logged(tmp_path):
"""Test that PII (code content, tokens, files) is not logged."""
from mcp_forge.security.audit import AuditLogger, AuditEventType, AuditSeverity
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
# Log execution request - should NOT include actual code
logger.log(
event_type=AuditEventType.EXECUTION_REQUEST,
severity=AuditSeverity.INFO,
message="Code execution requested",
details={
"code_hash": "abc123def456", # Hash is OK
# "code": "print('hello')" # Should NOT be logged
}
)
content = log_file.read_text()
# Should not contain actual code
assert "print" not in content
assert "hello" not in content
# Should contain hash
assert "abc123def456" in content
def test_log_container_operation(tmp_path):
"""Test log_container_operation convenience method."""
from mcp_forge.security.audit import AuditLogger, AuditEventType
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
logger.log_container_operation(
operation="create",
container_id="container-123",
image="mcp-forge/python:3.11",
session_id="session-456"
)
lines = log_file.read_text().strip().split("\n")
entry = json.loads(lines[0])
assert entry["event_type"] == "container.create"
assert entry["container_id"] == "container-123"
assert entry["image"] == "mcp-forge/python:3.11"
assert entry["session_id"] == "session-456"
def test_log_container_operation_with_error(tmp_path):
"""Test logging container operation with error."""
from mcp_forge.security.audit import AuditLogger
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
logger.log_container_operation(
operation="start",
container_id="container-123",
image="mcp-forge/python:3.11",
session_id="session-456",
error="Container not found"
)
lines = log_file.read_text().strip().split("\n")
entry = json.loads(lines[0])
assert "error" in entry
assert entry["error"] == "Container not found"
assert entry["severity"] == "error"
def test_log_creates_directory_if_not_exists(tmp_path):
"""Test that logger creates log directory if it doesn't exist."""
from mcp_forge.security.audit import AuditLogger, AuditEventType, AuditSeverity
log_dir = tmp_path / "nested" / "log" / "dir"
log_file = log_dir / "audit.log"
logger = AuditLogger(log_file)
logger.log(
event_type=AuditEventType.SESSION_CREATE,
severity=AuditSeverity.INFO,
message="Test"
)
assert log_file.exists()
assert log_file.parent.exists()
def test_audit_event_types():
"""Test that all required audit event types are defined."""
from mcp_forge.security.audit import AuditEventType
# Required event types from todo.md
assert hasattr(AuditEventType, "CONTAINER_CREATE")
assert hasattr(AuditEventType, "CONTAINER_START")
assert hasattr(AuditEventType, "CONTAINER_STOP")
assert hasattr(AuditEventType, "CONTAINER_REMOVE")
assert hasattr(AuditEventType, "EXECUTION_REQUEST")
assert hasattr(AuditEventType, "SECURITY_VIOLATION")
assert hasattr(AuditEventType, "BUILD_REQUEST")
assert hasattr(AuditEventType, "BUILD_COMPLETE")
assert hasattr(AuditEventType, "SESSION_CREATE")
assert hasattr(AuditEventType, "SESSION_DESTROY")
def test_audit_severity_levels():
"""Test that all required severity levels are defined."""
from mcp_forge.security.audit import AuditSeverity
assert hasattr(AuditSeverity, "INFO")
assert hasattr(AuditSeverity, "WARNING")
assert hasattr(AuditSeverity, "ERROR")
assert hasattr(AuditSeverity, "CRITICAL")
def test_log_with_all_optional_parameters(tmp_path):
"""Test logging with all optional parameters provided."""
from mcp_forge.security.audit import AuditLogger, AuditEventType, AuditSeverity
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
logger.log(
event_type=AuditEventType.BUILD_COMPLETE,
severity=AuditSeverity.INFO,
message="Build completed successfully",
session_id="session-123",
user_id="user-456",
details={"image": "custom-env", "duration": 120},
error=None
)
lines = log_file.read_text().strip().split("\n")
entry = json.loads(lines[0])
assert entry["session_id"] == "session-123"
assert entry["user_id"] == "user-456"
assert entry["details"]["image"] == "custom-env"
assert entry["details"]["duration"] == 120
def test_multiple_log_entries_on_separate_lines(tmp_path):
"""Test that multiple log entries are written on separate lines."""
from mcp_forge.security.audit import AuditLogger, AuditEventType, AuditSeverity
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
for i in range(5):
logger.log(
event_type=AuditEventType.CONTAINER_CREATE,
severity=AuditSeverity.INFO,
message=f"Entry {i}"
)
lines = log_file.read_text().strip().split("\n")
assert len(lines) == 5
# Each line should be parseable
for line in lines:
json.loads(line)
def test_log_security_violation_parameters(tmp_path):
"""Test log_security_violation includes all necessary information."""
from mcp_forge.security.audit import AuditLogger
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
logger.log_security_violation(
operation="volume_mount",
reason="Attempted to mount /etc",
session_id="session-789"
)
lines = log_file.read_text().strip().split("\n")
entry = json.loads(lines[0])
assert entry["event_type"] == "security.violation"
assert entry["severity"] == "critical"
assert entry["operation"] == "volume_mount"
assert entry["reason"] == "Attempted to mount /etc"
assert entry["session_id"] == "session-789"
def test_details_can_be_none(tmp_path):
"""Test that details parameter can be None."""
from mcp_forge.security.audit import AuditLogger, AuditEventType, AuditSeverity
log_file = tmp_path / "audit.log"
logger = AuditLogger(log_file)
logger.log(
event_type=AuditEventType.SESSION_CREATE,
severity=AuditSeverity.INFO,
message="Session created",
details=None
)
lines = log_file.read_text().strip().split("\n")
entry = json.loads(lines[0])
# Should work without error
assert "message" in entry

276
tests/security/test_resource_limits.py Normal file
View file

@ -0,0 +1,276 @@
"""
Tests for resource limits module.
Following TDD approach - these tests are written before implementation.
Tests cover all parsing and validation requirements from todo.md section 1.2.1.
"""
import pytest
def test_parse_memory_string_megabytes():
"""Test parsing memory string with megabytes suffix."""
from mcp_forge.security.resource_limits import parse_memory_string
result = parse_memory_string("512m")
assert result == 536870912 # 512 * 1024 * 1024
def test_parse_memory_string_gigabytes():
"""Test parsing memory string with gigabytes suffix."""
from mcp_forge.security.resource_limits import parse_memory_string
result = parse_memory_string("2g")
assert result == 2147483648 # 2 * 1024 * 1024 * 1024
def test_parse_memory_string_kilobytes():
"""Test parsing memory string with kilobytes suffix."""
from mcp_forge.security.resource_limits import parse_memory_string
result = parse_memory_string("1024k")
assert result == 1048576 # 1024 * 1024
def test_parse_memory_string_case_insensitive():
"""Test that memory string parsing is case-insensitive."""
from mcp_forge.security.resource_limits import parse_memory_string
assert parse_memory_string("512M") == 536870912
assert parse_memory_string("2G") == 2147483648
assert parse_memory_string("1024K") == 1048576
def test_parse_memory_string_invalid_format_raises_value_error():
"""Test that invalid format raises ValueError."""
from mcp_forge.security.resource_limits import parse_memory_string
with pytest.raises(ValueError) as exc_info:
parse_memory_string("invalid")
assert "invalid" in str(exc_info.value).lower() or "format" in str(exc_info.value).lower()
with pytest.raises(ValueError):
parse_memory_string("512x") # Invalid suffix
with pytest.raises(ValueError):
parse_memory_string("abc") # Not a number
def test_parse_memory_string_negative_value_raises_value_error():
"""Test that negative values raise ValueError."""
from mcp_forge.security.resource_limits import parse_memory_string
with pytest.raises(ValueError) as exc_info:
parse_memory_string("-512m")
assert "positive" in str(exc_info.value).lower() or "negative" in str(exc_info.value).lower()
def test_parse_memory_string_zero_value_raises_value_error():
"""Test that zero value raises ValueError."""
from mcp_forge.security.resource_limits import parse_memory_string
with pytest.raises(ValueError) as exc_info:
parse_memory_string("0m")
assert "positive" in str(exc_info.value).lower() or "zero" in str(exc_info.value).lower()
def test_parse_cpu_quota_valid_value():
"""Test that valid CPU quota values are accepted."""
from mcp_forge.security.resource_limits import parse_cpu_quota
result = parse_cpu_quota(50000)
assert result == 50000
result = parse_cpu_quota(100000) # 100% of one core
assert result == 100000
def test_parse_cpu_quota_max_limit():
"""Test that CPU quota has a reasonable maximum (10 cores)."""
from mcp_forge.security.resource_limits import parse_cpu_quota
# Should accept up to 1000000 (10 cores)
result = parse_cpu_quota(1000000)
assert result == 1000000
# Should reject more than 10 cores
with pytest.raises(ValueError) as exc_info:
parse_cpu_quota(1000001)
assert "1000000" in str(exc_info.value) or "maximum" in str(exc_info.value).lower()
def test_parse_cpu_quota_negative_raises_value_error():
"""Test that negative CPU quota raises ValueError."""
from mcp_forge.security.resource_limits import parse_cpu_quota
with pytest.raises(ValueError) as exc_info:
parse_cpu_quota(-1)
assert "positive" in str(exc_info.value).lower() or "negative" in str(exc_info.value).lower()
def test_parse_cpu_quota_zero_raises_value_error():
"""Test that zero CPU quota raises ValueError."""
from mcp_forge.security.resource_limits import parse_cpu_quota
with pytest.raises(ValueError) as exc_info:
parse_cpu_quota(0)
assert "positive" in str(exc_info.value).lower() or "zero" in str(exc_info.value).lower()
def test_parse_storage_string_same_as_memory():
"""Test that storage parsing works the same as memory parsing."""
from mcp_forge.security.resource_limits import parse_storage_string
assert parse_storage_string("1g") == 1073741824
assert parse_storage_string("512m") == 536870912
assert parse_storage_string("2048k") == 2097152
def test_resource_limits_class_initialization():
"""Test ResourceLimits class initializes correctly."""
from mcp_forge.security.resource_limits import ResourceLimits
limits = ResourceLimits(
memory="512m",
storage="1g",
cpu_quota=50000,
timeout=300
)
assert limits.memory_bytes == 536870912
assert limits.storage_bytes == 1073741824
assert limits.cpu_quota == 50000
assert limits.timeout == 300
def test_resource_limits_validates_memory():
"""Test that ResourceLimits validates memory string."""
from mcp_forge.security.resource_limits import ResourceLimits
with pytest.raises(ValueError):
ResourceLimits(
memory="invalid",
storage="1g",
cpu_quota=50000,
timeout=300
)
def test_resource_limits_validates_storage():
"""Test that ResourceLimits validates storage string."""
from mcp_forge.security.resource_limits import ResourceLimits
with pytest.raises(ValueError):
ResourceLimits(
memory="512m",
storage="invalid",
cpu_quota=50000,
timeout=300
)
def test_resource_limits_validates_cpu_quota():
"""Test that ResourceLimits validates CPU quota."""
from mcp_forge.security.resource_limits import ResourceLimits
with pytest.raises(ValueError):
ResourceLimits(
memory="512m",
storage="1g",
cpu_quota=-1,
timeout=300
)
def test_resource_limits_to_podman_params():
"""Test conversion to Podman container parameters."""
from mcp_forge.security.resource_limits import ResourceLimits
limits = ResourceLimits(
memory="512m",
storage="1g",
cpu_quota=50000,
timeout=300
)
params = limits.to_podman_params()
assert isinstance(params, dict)
assert "mem_limit" in params
assert params["mem_limit"] == "536870912" # Should be string for Podman
# CPU quota is set via cpu_quota parameter
assert "cpu_quota" in params
assert params["cpu_quota"] == 50000
def test_resource_limits_default_timeout():
"""Test that ResourceLimits has a default timeout."""
from mcp_forge.security.resource_limits import ResourceLimits
limits = ResourceLimits(
memory="512m",
storage="1g",
cpu_quota=50000
)
assert limits.timeout == 300 # Default from signature
def test_parse_memory_string_with_spaces():
"""Test parsing memory strings that have spaces."""
from mcp_forge.security.resource_limits import parse_memory_string
# Should handle spaces gracefully (strip them)
result = parse_memory_string(" 512m ")
assert result == 536870912
def test_parse_memory_string_bytes_suffix():
"""Test parsing memory string with bytes suffix (no multiplier)."""
from mcp_forge.security.resource_limits import parse_memory_string
# Just a number (bytes) - should this be supported?
# Based on architecture, we support k, m, g suffixes
# Plain numbers should probably raise an error for safety
with pytest.raises(ValueError):
parse_memory_string("1024")
def test_resource_limits_storage_quota_in_podman_params():
"""Test that storage limits are included in Podman params."""
from mcp_forge.security.resource_limits import ResourceLimits
limits = ResourceLimits(
memory="512m",
storage="1g",
cpu_quota=50000,
timeout=300
)
params = limits.to_podman_params()
# Storage limit might be set via storage_opt or similar
# The exact parameter depends on Podman API
assert "storage_bytes" in params or "storage_opt" in params
def test_cpu_quota_explanation():
"""Test that CPU quota values have clear meaning."""
from mcp_forge.security.resource_limits import parse_cpu_quota
# 100000 = 100% of one CPU core
# 50000 = 50% of one CPU core
# 200000 = 200% = 2 CPU cores
assert parse_cpu_quota(50000) == 50000 # 0.5 cores
assert parse_cpu_quota(100000) == 100000 # 1 core
assert parse_cpu_quota(200000) == 200000 # 2 cores
def test_parse_memory_with_decimal():
"""Test parsing memory strings with decimal values."""
from mcp_forge.security.resource_limits import parse_memory_string
# Should handle decimals
result = parse_memory_string("1.5g")
assert result == 1610612736 # 1.5 * 1024 * 1024 * 1024