Coverage for src / mcp_server_langgraph / execution / resource_limits.py: 98%
84 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-03 00:43 +0000
1"""
2Resource limits configuration for code execution
4Defines CPU, memory, timeout, and network constraints for sandboxed execution.
5Immutable configuration with validation and preset profiles.
6"""
8from dataclasses import dataclass, field
9from typing import Any, Literal
12class ResourceLimitError(Exception):
13 """Raised when resource limit validation fails"""
16NetworkMode = Literal["none", "allowlist", "unrestricted"]
19@dataclass(frozen=True)
20class ResourceLimits:
21 """
22 Resource limits for sandboxed code execution.
24 Immutable configuration ensuring consistent resource constraints.
25 All limits are validated on creation to prevent invalid configurations.
27 Example:
28 >>> limits = ResourceLimits(timeout_seconds=30, memory_limit_mb=512)
29 >>> assert limits.timeout_seconds == 30
31 Attributes:
32 timeout_seconds: Maximum execution time (1-600 seconds, default: 30)
33 memory_limit_mb: Maximum memory usage in MB (64-8192 MB, default: 512)
34 cpu_quota: CPU cores quota (0.1-8.0, default: 1.0)
35 disk_quota_mb: Maximum disk usage in MB (1-10240 MB, default: 100)
36 max_processes: Maximum number of processes (1-100, default: 1)
37 network_mode: Network access mode (none/allowlist/unrestricted, default: none)
38 allowed_domains: Domains allowed in 'allowlist' mode (default: empty)
39 """
41 timeout_seconds: int = 30
42 memory_limit_mb: int = 512
43 cpu_quota: float = 1.0
44 disk_quota_mb: int = 100
45 max_processes: int = 1
46 network_mode: NetworkMode = "none"
47 allowed_domains: tuple[str, ...] = field(default_factory=tuple)
49 def __post_init__(self) -> None:
50 """Validate resource limits after initialization"""
51 self._validate_timeout()
52 self._validate_memory()
53 self._validate_cpu()
54 self._validate_disk()
55 self._validate_processes()
56 self._validate_network()
58 def _validate_timeout(self) -> None:
59 """Validate timeout constraints"""
60 if self.timeout_seconds <= 0:
61 msg = f"Timeout must be positive, got {self.timeout_seconds}"
62 raise ResourceLimitError(msg)
64 if self.timeout_seconds > 600: # 10 minutes max
65 msg = f"Timeout cannot exceed 600 seconds, got {self.timeout_seconds}"
66 raise ResourceLimitError(msg)
68 def _validate_memory(self) -> None:
69 """Validate memory constraints"""
70 if self.memory_limit_mb <= 0:
71 msg = f"Memory limit must be positive, got {self.memory_limit_mb}"
72 raise ResourceLimitError(msg)
74 if self.memory_limit_mb < 64:
75 msg = f"Memory limit cannot be less than 64MB, got {self.memory_limit_mb}"
76 raise ResourceLimitError(msg)
78 if self.memory_limit_mb > 16384: # 16GB max
79 msg = f"Memory limit cannot exceed 16GB (16384MB), got {self.memory_limit_mb}"
80 raise ResourceLimitError(msg)
82 def _validate_cpu(self) -> None:
83 """Validate CPU quota constraints"""
84 if self.cpu_quota <= 0:
85 msg = f"CPU quota must be positive, got {self.cpu_quota}"
86 raise ResourceLimitError(msg)
88 if self.cpu_quota > 8.0:
89 msg = f"CPU quota cannot exceed 8.0 cores, got {self.cpu_quota}"
90 raise ResourceLimitError(msg)
92 def _validate_disk(self) -> None:
93 """Validate disk quota constraints"""
94 if self.disk_quota_mb <= 0:
95 msg = f"Disk quota must be positive, got {self.disk_quota_mb}"
96 raise ResourceLimitError(msg)
98 if self.disk_quota_mb > 10240: # 10GB max
99 msg = f"Disk quota cannot exceed 10GB (10240MB), got {self.disk_quota_mb}"
100 raise ResourceLimitError(msg)
102 def _validate_processes(self) -> None:
103 """Validate max processes constraints"""
104 if self.max_processes <= 0:
105 msg = f"Max processes must be positive, got {self.max_processes}"
106 raise ResourceLimitError(msg)
108 if self.max_processes > 100:
109 msg = f"Max processes cannot exceed 100, got {self.max_processes}"
110 raise ResourceLimitError(msg)
112 def _validate_network(self) -> None:
113 """Validate network configuration"""
114 valid_modes: tuple[NetworkMode, ...] = ("none", "allowlist", "unrestricted")
115 if self.network_mode not in valid_modes:
116 msg = f"Network mode must be one of {valid_modes}, got '{self.network_mode}'"
117 raise ResourceLimitError(msg)
119 # Note: allowlist with empty domains is allowed (effectively blocks all network)
120 # This allows configuring the profile first, then adding domains later
122 def to_dict(self) -> dict[str, Any]:
123 """Convert resource limits to dictionary"""
124 return {
125 "timeout_seconds": self.timeout_seconds,
126 "memory_limit_mb": self.memory_limit_mb,
127 "cpu_quota": self.cpu_quota,
128 "disk_quota_mb": self.disk_quota_mb,
129 "max_processes": self.max_processes,
130 "network_mode": self.network_mode,
131 "allowed_domains": list(self.allowed_domains),
132 }
134 @classmethod
135 def from_dict(cls, data: dict[str, Any]) -> "ResourceLimits":
136 """Create resource limits from dictionary"""
137 # Convert allowed_domains list to tuple for immutability
138 if "allowed_domains" in data and isinstance(data["allowed_domains"], list): 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true
139 data = {**data, "allowed_domains": tuple(data["allowed_domains"])}
141 return cls(**data)
143 def is_within(self, other: "ResourceLimits") -> bool:
144 """
145 Check if these limits are within (stricter than or equal to) another set of limits.
147 Args:
148 other: Resource limits to compare against
150 Returns:
151 True if all limits are within the other limits
152 """
153 return (
154 self.timeout_seconds <= other.timeout_seconds
155 and self.memory_limit_mb <= other.memory_limit_mb
156 and self.cpu_quota <= other.cpu_quota
157 and self.disk_quota_mb <= other.disk_quota_mb
158 and self.max_processes <= other.max_processes
159 )
161 @classmethod
162 def development(cls) -> "ResourceLimits":
163 """
164 Development profile with relaxed limits for local testing.
166 Returns:
167 ResourceLimits configured for development
168 """
169 return cls(
170 timeout_seconds=300, # 5 minutes
171 memory_limit_mb=2048, # 2GB
172 cpu_quota=2.0, # 2 CPUs
173 disk_quota_mb=1024, # 1GB
174 max_processes=10,
175 network_mode="unrestricted", # Full network for development
176 allowed_domains=(),
177 )
179 @classmethod
180 def production(cls) -> "ResourceLimits":
181 """
182 Production profile with conservative limits for security.
184 SECURITY (OpenAI Codex Finding #3):
185 Network access is DISABLED (network_mode="none") because domain allowlisting
186 is not yet implemented. Using network_mode="allowlist" would be misleading
187 since it currently fails closed to "none" anyway.
189 To enable network access in production, explicitly set network_mode="unrestricted"
190 and understand the security implications.
192 Returns:
193 ResourceLimits configured for production
194 """
195 return cls(
196 timeout_seconds=30, # 30 seconds
197 memory_limit_mb=512, # 512MB
198 cpu_quota=1.0, # 1 CPU
199 disk_quota_mb=100, # 100MB
200 max_processes=1,
201 network_mode="none", # SECURITY: Network disabled (allowlist not implemented)
202 allowed_domains=(), # Reserved for future allowlist implementation
203 )
205 @classmethod
206 def testing(cls) -> "ResourceLimits":
207 """
208 Testing profile with minimal limits for fast execution.
210 Returns:
211 ResourceLimits configured for testing
212 """
213 return cls(
214 timeout_seconds=10, # 10 seconds
215 memory_limit_mb=256, # 256MB
216 cpu_quota=0.5, # 0.5 CPU
217 disk_quota_mb=50, # 50MB
218 max_processes=1,
219 network_mode="none", # No network for tests
220 allowed_domains=(),
221 )
223 @classmethod
224 def data_processing(cls) -> "ResourceLimits":
225 """
226 Data processing profile with higher memory and CPU for analytics.
228 Returns:
229 ResourceLimits configured for data processing
230 """
231 return cls(
232 timeout_seconds=300, # 5 minutes
233 memory_limit_mb=4096, # 4GB
234 cpu_quota=4.0, # 4 CPUs
235 disk_quota_mb=512, # 512MB
236 max_processes=4,
237 network_mode="allowlist",
238 allowed_domains=(), # Configure per use case
239 )