Coverage for src/taipanstack/security/decorators.py: 100%

1""" 

2Security decorators for robust Python applications. 

3 

4Provides decorators for input validation, exception handling, 

5timeout control, and other security patterns. Compatible with 

6any Python framework (Flask, FastAPI, Django, etc.). 

7""" 

8 

9import functools 

10import inspect 

11import math 

12import signal 

13import sys 

14import threading 

15from collections.abc import Callable 

16from types import FrameType 

17from typing import ParamSpec, TypeVar 

18 

19from taipanstack.security.guards import SecurityError 

20 

21P = ParamSpec("P") 

22R = TypeVar("R") 

23T = TypeVar("T") 

24 

25 

26def _apply_single_validator( 

27 param_name: str, 

28 value: object, 

29 validator: Callable[[object], object], 

30) -> object | None: 

31 try: 

32 return validator(value) 

33 except (ValueError, TypeError) as e: 

34 raise ValidationError( 

35 str(e), 

36 param_name=param_name, 

37 value=repr(value)[:100], 

38 ) from e 

39 

40 

41def _validate_bound_arguments( 

42 bound: inspect.BoundArguments, 

43 validators: dict[str, Callable[[object], object]], 

44) -> None: 

45 for param_name, validator in validators.items(): 

46 if param_name not in bound.arguments: # type: ignore[misc] 

47 continue 

48 value = bound.arguments[param_name] # type: ignore[misc] 

49 validated = _apply_single_validator(param_name, value, validator) # type: ignore[misc] 

50 if validated is not None: 

51 bound.arguments[param_name] = validated # type: ignore[misc] 

52 

53 

54def _handle_guarded_exception( 

55 e: Exception, 

56 func_name: str, 

57 reraise_as: type[Exception] | None, 

58 log_errors: bool, 

59) -> None: 

60 if log_errors: 

61 import logging 

62 

63 logging.getLogger("taipanstack.security").warning( 

64 "Exception caught in %s: %s", 

65 func_name, 

66 str(e), 

67 ) 

68 

69 if reraise_as is not None: 

70 if reraise_as == SecurityError: 

71 raise SecurityError( 

72 str(e), 

73 guard_name="guard_exceptions", 

74 ) from e 

75 raise reraise_as(str(e)) from e 

76 

77 

78def _check_bound_types( 

79 bound: inspect.BoundArguments, 

80 type_hints: dict[str, type], 

81) -> None: 

82 for param_name, expected_type in type_hints.items(): 

83 if param_name in bound.arguments: # type: ignore[misc] 

84 value = bound.arguments[param_name] # type: ignore[misc] 

85 if not isinstance(value, expected_type): # type: ignore[misc] 

86 raise TypeError( 

87 f"Parameter '{param_name}' expected " 

88 f"{expected_type.__name__}, got {type(value).__name__}", # type: ignore[misc] 

89 ) 

90 

91 

92class OperationTimeoutError(Exception): 

93 """Raised when a function exceeds its timeout limit.""" 

94 

95 def __init__(self, seconds: float, func_name: str = "function") -> None: 

96 """Initialize OperationTimeoutError. 

97 

98 Args: 

99 seconds: The timeout that was exceeded. 

100 func_name: Name of the function that timed out. 

101 

102 """ 

103 self.seconds = seconds 

104 self.func_name = func_name 

105 super().__init__(f"{func_name} timed out after {seconds} seconds") 

106 

107 

108class ValidationError(Exception): 

109 """Raised when input validation fails.""" 

110 

111 def __init__( 

112 self, 

113 message: str, 

114 param_name: str | None = None, 

115 value: object = None, 

116 ) -> None: 

117 """Initialize ValidationError. 

118 

119 Args: 

120 message: Description of the validation failure. 

121 param_name: Name of the parameter that failed. 

122 value: The invalid value (sanitized). 

123 

124 """ 

125 self.param_name = param_name 

126 self.value = value 

127 super().__init__(message) 

128 

129 

130def validate_inputs( 

131 **validators: Callable[[object], object], 

132) -> Callable[[Callable[P, R]], Callable[P, R]]: 

133 """Decorator to validate function inputs. 

134 

135 Validates function arguments using provided validator functions. 

136 Validators should raise ValueError or ValidationError on invalid input. 

137 

138 Args: 

139 **validators: Mapping of parameter names to validator functions. 

140 

141 Returns: 

142 Decorated function with input validation. 

143 

144 Example: 

145 >>> from taipanstack.security.validators import validate_email, validate_port 

146 >>> @validate_inputs(email=validate_email, port=validate_port) 

147 ... def connect(email: str, port: int) -> None: 

148 ... pass 

149 >>> connect(email="invalid", port=8080) 

150 ValidationError: Invalid email format: invalid 

151 

152 """ 

153 

154 def decorator(func: Callable[P, R]) -> Callable[P, R]: 

155 sig = inspect.signature(func) 

156 

157 @functools.wraps(func) 

158 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

159 # Bind and apply defaults on each call 

160 bound = sig.bind(*args, **kwargs) 

161 bound.apply_defaults() 

162 

163 # Validate each parameter that has a validator 

164 _validate_bound_arguments(bound, validators) 

165 

166 # Call original function with validated arguments 

167 return func(*bound.args, **bound.kwargs) # type: ignore[misc] 

168 

169 return wrapper 

170 

171 return decorator 

172 

173 

174def guard_exceptions( 

175 *, 

176 catch: tuple[type[Exception], ...] = (Exception,), 

177 reraise_as: type[Exception] | None = None, 

178 default: T | None = None, 

179 log_errors: bool = True, 

180) -> Callable[[Callable[P, R]], Callable[P, R | T | None]]: 

181 """Decorator to safely handle exceptions. 

182 

183 Catches exceptions and optionally re-raises as a different type 

184 or returns a default value. 

185 

186 Args: 

187 catch: Exception types to catch. 

188 reraise_as: Exception type to re-raise as (None = don't reraise). 

189 default: Default value to return if exception caught and not reraised. 

190 log_errors: Whether to log caught exceptions. 

191 

192 Returns: 

193 Decorated function with exception handling. 

194 

195 Example: 

196 >>> @guard_exceptions(catch=(IOError,), reraise_as=SecurityError) 

197 ... def read_file(path: str) -> str: 

198 ... return open(path).read() 

199 >>> read_file("/nonexistent") 

200 SecurityError: [guard_exceptions] ... 

201 

202 """ 

203 

204 def decorator(func: Callable[P, R]) -> Callable[P, R | T | None]: 

205 @functools.wraps(func) 

206 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R | T | None: 

207 try: 

208 return func(*args, **kwargs) 

209 except catch as e: 

210 _handle_guarded_exception(e, func.__name__, reraise_as, log_errors) 

211 return default 

212 

213 return wrapper 

214 

215 return decorator 

216 

217 

218def _validate_timeout_seconds(seconds: float) -> None: 

219 """Validate timeout duration.""" 

220 if ( 

221 not isinstance(seconds, (int, float)) 

222 or not math.isfinite(seconds) 

223 or seconds < 0 

224 ): 

225 raise ValueError("timeout must be a finite non-negative number") 

226 

227 

228def _is_signal_supported(use_signal: bool) -> bool: 

229 """Check if signal-based timeout can be used.""" 

230 return ( 

231 use_signal 

232 and sys.platform != "win32" 

233 and threading.current_thread() is threading.main_thread() 

234 ) 

235 

236 

237def timeout( 

238 seconds: float, 

239 *, 

240 use_signal: bool = True, 

241) -> Callable[[Callable[P, R]], Callable[P, R]]: 

242 """Decorator to limit function execution time. 

243 

244 Uses signal-based timeout on Unix or thread-based on Windows. 

245 Signal-based is more reliable but only works in main thread. 

246 

247 Args: 

248 seconds: Maximum execution time in seconds. 

249 use_signal: Use signal-based timeout (Unix only, main thread only). 

250 

251 Returns: 

252 Decorated function with timeout. 

253 

254 Example: 

255 >>> @timeout(5.0) 

256 ... def slow_operation() -> str: 

257 ... import time 

258 ... time.sleep(10) 

259 ... return "done" 

260 >>> slow_operation() 

261 TimeoutError: slow_operation timed out after 5.0 seconds 

262 

263 """ 

264 # Security Enhancement: explicitly validate bounds using math.isfinite() 

265 # and check for non-negative limits to prevent silent NaN propagation, 

266 # unhandled ValueError exceptions from threading/asyncio primitives, 

267 # or unexpected infinite blocking behaviors. 

268 _validate_timeout_seconds(seconds) 

269 

270 def decorator(func: Callable[P, R]) -> Callable[P, R]: 

271 @functools.wraps(func) 

272 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

273 if _is_signal_supported(use_signal): 

274 return _timeout_with_signal( 

275 func, 

276 seconds, 

277 args, 

278 dict(kwargs), 

279 ) 

280 return _timeout_with_thread( 

281 func, 

282 seconds, 

283 args, 

284 dict(kwargs), 

285 ) 

286 

287 return wrapper 

288 

289 return decorator 

290 

291 

292def _timeout_with_signal( 

293 func: Callable[..., R], 

294 seconds: float, 

295 args: tuple[object, ...], 

296 kwargs: dict[str, object], 

297) -> R: 

298 """Implement timeout using Unix signals.""" 

299 

300 def handler(_signum: int, _frame: FrameType | None) -> None: 

301 raise OperationTimeoutError(seconds, func.__name__) # type: ignore[misc] 

302 

303 # Set up signal handler 

304 old_handler = signal.signal(signal.SIGALRM, handler) 

305 signal.setitimer(signal.ITIMER_REAL, seconds) 

306 

307 try: 

308 return func(*args, **kwargs) 

309 finally: 

310 # Restore old handler and cancel alarm 

311 signal.setitimer(signal.ITIMER_REAL, 0) 

312 signal.signal(signal.SIGALRM, old_handler) 

313 

314 

315def _check_thread_timeout( 

316 thread: threading.Thread, seconds: float, func_name: str 

317) -> None: 

318 """Raise OperationTimeoutError if the thread is still running.""" 

319 if thread.is_alive(): 

320 raise OperationTimeoutError(seconds, func_name) 

321 

322 

323def _check_thread_exception(exception: list[BaseException]) -> None: 

324 """Raise the first exception if the thread encountered any.""" 

325 if exception: 

326 raise exception[0] 

327 

328 

329def _timeout_with_thread( 

330 func: Callable[..., R], 

331 seconds: float, 

332 args: tuple[object, ...], 

333 kwargs: dict[str, object], 

334) -> R: 

335 """Implement timeout using a separate thread.""" 

336 result: list[R] = [] 

337 exception: list[BaseException] = [] 

338 

339 def target() -> None: 

340 try: 

341 result.append(func(*args, **kwargs)) 

342 except BaseException as e: 

343 exception.append(e) 

344 

345 thread = threading.Thread(target=target) 

346 thread.daemon = True 

347 thread.start() 

348 thread.join(timeout=seconds) 

349 

350 _check_thread_timeout(thread, seconds, func.__name__) # type: ignore[misc] 

351 _check_thread_exception(exception) 

352 

353 return result[0] 

354 

355 

356def _build_deprecation_message( 

357 func_name: str, message: str, removal_version: str | None 

358) -> str: 

359 """Build the deprecation warning message.""" 

360 msg = f"{func_name} is deprecated." 

361 if removal_version: 

362 msg += f" Will be removed in version {removal_version}." 

363 if message: 

364 msg += f" {message}" 

365 return msg 

366 

367 

368def deprecated( 

369 message: str = "", 

370 *, 

371 removal_version: str | None = None, 

372) -> Callable[[Callable[P, R]], Callable[P, R]]: 

373 """Mark a function as deprecated. 

374 

375 Emits a warning when the decorated function is called. 

376 

377 Args: 

378 message: Additional deprecation message. 

379 removal_version: Version when function will be removed. 

380 

381 Returns: 

382 Decorated function that warns on use. 

383 

384 Example: 

385 >>> @deprecated("Use new_function instead", removal_version="2.0") 

386 ... def old_function() -> None: 

387 ... pass 

388 

389 """ 

390 

391 def decorator(func: Callable[P, R]) -> Callable[P, R]: 

392 @functools.wraps(func) 

393 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

394 import warnings 

395 

396 msg = _build_deprecation_message(func.__name__, message, removal_version) 

397 

398 warnings.warn(msg, DeprecationWarning, stacklevel=2) 

399 return func(*args, **kwargs) 

400 

401 return wrapper 

402 

403 return decorator 

404 

405 

406def require_type( 

407 **type_hints: type, 

408) -> Callable[[Callable[P, R]], Callable[P, R]]: 

409 """Decorator to enforce runtime type checking. 

410 

411 Validates that arguments match specified types at runtime. 

412 

413 Args: 

414 **type_hints: Mapping of parameter names to expected types. 

415 

416 Returns: 

417 Decorated function with type checking. 

418 

419 Example: 

420 >>> @require_type(name=str, count=int) 

421 ... def greet(name: str, count: int) -> None: 

422 ... print(f"Hello {name}" * count) 

423 >>> greet(name=123, count=2) 

424 TypeError: Parameter 'name' expected str, got int 

425 

426 """ 

427 

428 def decorator(func: Callable[P, R]) -> Callable[P, R]: 

429 sig = inspect.signature(func) 

430 

431 @functools.wraps(func) 

432 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

433 bound = sig.bind(*args, **kwargs) 

434 bound.apply_defaults() 

435 

436 _check_bound_types(bound, type_hints) 

437 

438 return func(*bound.args, **bound.kwargs) # type: ignore[misc] 

439 

440 return wrapper 

441 

442 return decorator