diff --git a/guard_core/_utils/detection_scan.py b/guard_core/_utils/detection_scan.py index 148cda27..9d98ac00 100644 --- a/guard_core/_utils/detection_scan.py +++ b/guard_core/_utils/detection_scan.py @@ -2,9 +2,9 @@ import logging import re import time -from collections.abc import Callable, Iterator +from collections.abc import Awaitable, Callable, Iterator from contextlib import contextmanager -from typing import Any +from typing import Any, TypeVar from guard_core._utils.detection_config import ( _DEFAULT_BINARY_MIN_RUN_LENGTH, @@ -14,9 +14,12 @@ _DEFAULT_MAX_SCAN_VALUES, ) from guard_core._utils.logging_utils import _log_at_level, _sanitize_for_reporting +from guard_core._utils.pattern_source_redaction import _redact_pattern_source logger = logging.getLogger("guard_core") +_T = TypeVar("_T") + _scanned_value_count: contextvars.ContextVar[int] = contextvars.ContextVar( "guard_core_detection_scanned_value_count", default=0 ) @@ -50,6 +53,12 @@ _scan_time_spent: contextvars.ContextVar[float] = contextvars.ContextVar( "guard_core_detection_scan_time_spent", default=0.0 ) +_scan_cpu_mark: contextvars.ContextVar[float | None] = contextvars.ContextVar( + "guard_core_detection_scan_cpu_mark", default=None +) +_scan_slice_start: contextvars.ContextVar[float | None] = contextvars.ContextVar( + "guard_core_detection_scan_slice_start", default=None +) _semantic_excluded_headers: contextvars.ContextVar[frozenset[str]] = ( contextvars.ContextVar( "guard_core_detection_semantic_excluded_headers", default=frozenset() @@ -77,6 +86,7 @@ def _scan_value_budget( time_cap_token = _scan_time_cap.set(max_scan_seconds) time_spent_token = _scan_time_spent.set(0.0) time_warned_token = _scan_time_cap_warned.set(False) + slice_token = _scan_slice_start.set(time.monotonic()) semantic_token = _semantic_excluded_headers.set(semantic_excluded_headers) try: yield @@ -92,6 +102,7 @@ def _scan_value_budget( _scan_time_cap.reset(time_cap_token) _scan_time_spent.reset(time_spent_token) _scan_time_cap_warned.reset(time_warned_token) + _scan_slice_start.reset(slice_token) _semantic_excluded_headers.reset(semantic_token) @@ -128,13 +139,36 @@ def _record_scan_time(elapsed: float) -> None: _scan_time_spent.set(_scan_time_spent.get() + elapsed) -@contextmanager -def _record_inline_cpu() -> Iterator[None]: - start = _thread_cpu_seconds() +def _open_cpu_pass() -> contextvars.Token[float | None]: + return _scan_cpu_mark.set( + None if _scan_time_cap.get() is None else _thread_cpu_seconds() + ) + + +def _close_cpu_pass(token: contextvars.Token[float | None]) -> None: + _charge_inline_cpu() + _scan_cpu_mark.reset(token) + + +def _charge_inline_cpu() -> None: + mark = _scan_cpu_mark.get() + if mark is not None: + now = _thread_cpu_seconds() + _record_scan_time(now - mark) + _scan_cpu_mark.set(now) + + +def _restart_inline_cpu() -> None: + if _scan_cpu_mark.get() is not None: + _scan_cpu_mark.set(_thread_cpu_seconds()) + + +async def _suspended(awaitable: Awaitable[_T]) -> _T: + _charge_inline_cpu() try: - yield + return await awaitable finally: - _record_scan_time(_thread_cpu_seconds() - start) + _restart_inline_cpu() def _timed_in_worker( @@ -233,53 +267,6 @@ def _warn_json_depth_cap_reached_once(client_ip: str) -> None: ) -_PATTERN_SOURCE_NON_TOKEN_RE = re.compile(r"[^a-z0-9_-]+") -_PATTERN_SOURCE_ESCAPE_RE = re.compile( - r"\\(?:[xX]([0-9A-Fa-f]{2})|u([0-9A-Fa-f]{4})|U([0-9A-Fa-f]{8})|([0-7]{1,3}))" -) - - -def _decode_pattern_source_escape(match: re.Match[str]) -> str: - hex_body = match.group(1) or match.group(2) or match.group(3) - code = int(hex_body, 16) if hex_body else int(match.group(4), 8) - return chr(code) if code <= 0x10FFFF else match.group(0) - - -def _pattern_source_names_a_sensitive_field( - pattern_source: str, sensitive_names: frozenset[str] -) -> bool: - decoded = _PATTERN_SOURCE_ESCAPE_RE.sub( - _decode_pattern_source_escape, pattern_source - ) - normalized = _PATTERN_SOURCE_NON_TOKEN_RE.sub("", decoded.lower()) - return any(name in normalized for name in sensitive_names) - - -def _redact_pattern_source(pattern_source: str) -> str: - from guard_core._utils.request_logging import ( - _merge_sensitive_log_body_fields, - _merged_sensitive_names, - redact_blob_for_display, - ) - from guard_core.handlers.suspatterns_handler import sus_patterns_handler - - sensitive_params = sus_patterns_handler._sensitive_params_union - sensitive_body_fields = sus_patterns_handler._sensitive_body_fields_union - sensitive_headers = sus_patterns_handler._sensitive_headers_union - - sensitive_names = _merged_sensitive_names( - sensitive_params, - _merge_sensitive_log_body_fields(sensitive_body_fields), - sensitive_headers, - ) - if _pattern_source_names_a_sensitive_field(pattern_source, sensitive_names): - return "[REDACTED]" - - return redact_blob_for_display( - pattern_source, sensitive_params, sensitive_body_fields, sensitive_headers - ) - - def _build_threat_message(threat: dict[str, Any]) -> str: if threat["type"] == "regex": return f"Value matched pattern '{_redact_pattern_source(threat['pattern'])}'" @@ -331,31 +318,36 @@ async def _fallback_pattern_check( normalized_context = sus_patterns_handler._normalize_context(context) all_compiled = await sus_patterns_handler.get_all_compiled_patterns() binary_prefix = build_binary_prefix(value) - for pattern, _contexts, category in all_compiled: - if _scan_time_budget_exhausted(): - return True, "Scan time budget exhausted (fallback)" - pattern_start = time.monotonic() - try: - threat, _timeout_occurred = await sus_patterns_handler._check_regex_pattern( - pattern, - value, - client_ip, - pattern_start, - category, - context=normalized_context, - binary_prefix=binary_prefix, - ) - except RecursionError: - logger.warning( - "Fallback pattern search hit the regex engine's recursion " - "limit on pattern %r; skipping it and continuing the " - "fallback scan", - _redact_pattern_source(pattern.pattern), - ) - continue - if threat: - return True, "Value matched pattern (fallback)" - return False, "" + mark_token = _open_cpu_pass() + try: + for pattern, _contexts, category in all_compiled: + _charge_inline_cpu() + if _scan_time_budget_exhausted(): + return True, "Scan time budget exhausted (fallback)" + pattern_start = time.monotonic() + try: + threat, _ = await sus_patterns_handler._check_regex_pattern( + pattern, + value, + client_ip, + pattern_start, + category, + context=normalized_context, + binary_prefix=binary_prefix, + ) + except RecursionError: + logger.warning( + "Fallback pattern search hit the regex engine's recursion " + "limit on pattern %r; skipping it and continuing the " + "fallback scan", + _redact_pattern_source(pattern.pattern), + ) + continue + if threat: + return True, "Value matched pattern (fallback)" + return False, "" + finally: + _close_cpu_pass(mark_token) def _resolve_json_redact_all( @@ -561,17 +553,18 @@ def _check_always_scan_header(value: str) -> tuple[bool, str, list[dict]]: async def _compiler_finditer( compiler: Any, pattern: re.Pattern, category: str, content: str ) -> list[re.Match]: - inline_scan = category != "custom" + if category != "custom": + inline_finder = compiler.create_async_safe_finditer_matcher( + pattern, inline_safe=True + ) + inline_matches: list[re.Match] = await inline_finder(content) + return inline_matches worker_cpu: list[float] = [] finder = compiler.create_async_safe_finditer_matcher( pattern, - inline_safe=inline_scan, + inline_safe=False, worker_wrapper=lambda work: _timed_in_worker(work, worker_cpu), ) - if inline_scan: - with _record_inline_cpu(): - matches: list[re.Match] = await finder(content) - return matches - matches = await finder(content) + matches: list[re.Match] = await _suspended(finder(content)) _record_scan_time(sum(worker_cpu) if worker_cpu else compiler.default_timeout) return matches diff --git a/guard_core/_utils/pattern_source_redaction.py b/guard_core/_utils/pattern_source_redaction.py new file mode 100644 index 00000000..131cfdc5 --- /dev/null +++ b/guard_core/_utils/pattern_source_redaction.py @@ -0,0 +1,47 @@ +import re + +_PATTERN_SOURCE_NON_TOKEN_RE = re.compile(r"[^a-z0-9_-]+") +_PATTERN_SOURCE_ESCAPE_RE = re.compile( + r"\\(?:[xX]([0-9A-Fa-f]{2})|u([0-9A-Fa-f]{4})|U([0-9A-Fa-f]{8})|([0-7]{1,3}))" +) + + +def _decode_pattern_source_escape(match: re.Match[str]) -> str: + hex_body = match.group(1) or match.group(2) or match.group(3) + code = int(hex_body, 16) if hex_body else int(match.group(4), 8) + return chr(code) if code <= 0x10FFFF else match.group(0) + + +def _pattern_source_names_a_sensitive_field( + pattern_source: str, sensitive_names: frozenset[str] +) -> bool: + decoded = _PATTERN_SOURCE_ESCAPE_RE.sub( + _decode_pattern_source_escape, pattern_source + ) + normalized = _PATTERN_SOURCE_NON_TOKEN_RE.sub("", decoded.lower()) + return any(name in normalized for name in sensitive_names) + + +def _redact_pattern_source(pattern_source: str) -> str: + from guard_core._utils.request_logging import ( + _merge_sensitive_log_body_fields, + _merged_sensitive_names, + redact_blob_for_display, + ) + from guard_core.handlers.suspatterns_handler import sus_patterns_handler + + sensitive_params = sus_patterns_handler._sensitive_params_union + sensitive_body_fields = sus_patterns_handler._sensitive_body_fields_union + sensitive_headers = sus_patterns_handler._sensitive_headers_union + + sensitive_names = _merged_sensitive_names( + sensitive_params, + _merge_sensitive_log_body_fields(sensitive_body_fields), + sensitive_headers, + ) + if _pattern_source_names_a_sensitive_field(pattern_source, sensitive_names): + return "[REDACTED]" + + return redact_blob_for_display( + pattern_source, sensitive_params, sensitive_body_fields, sensitive_headers + ) diff --git a/guard_core/detection_engine/monitor.py b/guard_core/detection_engine/monitor.py index 4389cdb7..6bbf9111 100644 --- a/guard_core/detection_engine/monitor.py +++ b/guard_core/detection_engine/monitor.py @@ -182,6 +182,8 @@ async def _notify_callbacks( agent_handler: Any, correlation_id: str | None, ) -> None: + if not self.anomaly_callbacks: + return safe_anomaly = self._sanitize_anomaly_data(anomaly) for callback in self.anomaly_callbacks: diff --git a/guard_core/detection_engine/monitor_anomalies.py b/guard_core/detection_engine/monitor_anomalies.py index 72784c78..a816434b 100644 --- a/guard_core/detection_engine/monitor_anomalies.py +++ b/guard_core/detection_engine/monitor_anomalies.py @@ -48,8 +48,9 @@ def detect_statistical_anomaly( if anomaly_threshold >= 0 and metric.execution_time <= avg_time: return None - variance = math.fsum((t - avg_time) ** 2 for t in recent_times) / (sample_count - 1) - std_time = variance**0.5 + std_time = math.dist(recent_times, [avg_time] * sample_count) / math.sqrt( + sample_count - 1 + ) if std_time <= 0: return None diff --git a/guard_core/handlers/_suspatterns_regex.py b/guard_core/handlers/_suspatterns_regex.py index 31ad1ef9..c4fef725 100644 --- a/guard_core/handlers/_suspatterns_regex.py +++ b/guard_core/handlers/_suspatterns_regex.py @@ -8,12 +8,18 @@ from guard_core._utils.detection_scan import ( _EVENT_LOOP_SLICE_SECONDS, _SCAN_TIME_BUDGET_EXHAUSTED_PATTERN, + _charge_inline_cpu, + _close_cpu_pass, _compiler_finditer, - _record_inline_cpu, + _open_cpu_pass, _record_scan_time, _redact_pattern_source, + _restart_inline_cpu, + _scan_slice_start, _scan_time_budget_exhausted, _scan_time_budget_exhausted_threat, + _scan_time_cap, + _suspended, _timed_in_worker, _warn_scan_time_cap_reached_once, ) @@ -28,7 +34,6 @@ report_scan_timeout, shared_regex_executor, ) -from guard_core.detection_engine.scan_window import bounded_finditer from guard_core.handlers._suspatterns_enhanced import _SusPatternsEnhancedMixin from guard_core.handlers._suspatterns_ldap_ipv4 import ( _LEGACY_IPV4_HOST_RE, @@ -54,6 +59,10 @@ _pickle_global_candidate_is_injection, ) from guard_core.handlers._suspatterns_registry import _SusPatternsRegistryMixin +from guard_core.handlers._suspatterns_scan_windows import ( + _SCAN_WINDOW_PATTERNS, + _iter_scan_window_matches, +) from guard_core.handlers._suspatterns_shell_sources import ( _BRACE_EXPANSION_COMMAND_RE, _GLOB_WILDCARD_ATOM_RE, @@ -98,8 +107,6 @@ from guard_core.handlers._suspatterns_state import _DetectionState from guard_core.handlers._suspatterns_xml_xxe import ( _XML_XXE_PUBLIC_EXTERNAL_DTD_COMPILED_RE, - _xml_internal_entity_finditer, - _xml_system_finditer, _xml_xxe_public_external_dtd_finditer, ) @@ -242,53 +249,6 @@ def _pattern_should_be_skipped( ), } -_SCAN_WINDOW_BOUND_SOURCES: dict[str, tuple[tuple[str, str], ...]] = { - r"]*>[^<]*<\/script\s*>": ((r""),), - ( - r"(?:<[A-Za-z/][^<>]*style\s*=\s{0,20}[\"']?[^<>\"']*" - r"(?:expression|behavior|url)\s*\([^)]*\))" - ): ((r"<[A-Za-z/][^<>]*style\s*=", r"\)"),), - r"(?:]*>[\s\S]*<\/object\s*>)": ((r""),), - r"(?:]*>[\s\S]*<\/embed\s*>)": ((r""),), - r"(?:]*>[\s\S]*<\/applet\s*>)": ((r""),), - r"\.\.;[^/\\]*[/\\]": ((r"\.\.;", r"[/\\]"),), - ( - r"=(?:https?|ftp):\/\/[^\s'\"<>]+\/[^\s'\"<>\/]*" - r"\.(?:phtml|php[3-5]?|phar|jsp|aspx?|pl|py|txt|inc)(?![a-zA-Z0-9])" - ): ( - ( - r"=(?:https?|ftp):\/\/", - r"\.(?:phtml|php\d*|phar|jsp|aspx?|pl|py|txt|inc)[a-zA-Z0-9]*", - ), - ), - r"]+SYSTEM[^>]+>": ((r""),), - r"(?:)": ((r""),), - r"\[]*\[[\s\S]*? Iterator[re.Match]: - if pattern.pattern == r"]+SYSTEM[^>]+>": - yield from _xml_system_finditer(content) - return - if pattern.pattern == r"\[]*\[[\s\S]*? str: return value.encode("utf-8", errors="surrogateescape").decode( @@ -348,16 +308,34 @@ def _first_accepted_regex_threat( context: str = "unknown", binary_prefix: list[int] | None = None, ) -> dict[str, Any] | None: - with _record_inline_cpu(): - for match in matches: - threat = _build_regex_threat( - pattern, match, category, pattern_start, context, binary_prefix - ) - if threat: - return threat + for match in matches: + threat = _build_regex_threat( + pattern, match, category, pattern_start, context, binary_prefix + ) + if threat: + return threat return None +def _pass_slice_start() -> tuple[float, bool]: + request_slice_start = _scan_slice_start.get() + if request_slice_start is None: + return time.monotonic(), False + return request_slice_start, True + + +async def _yield_once_slice_elapsed(slice_start: float, request_scoped: bool) -> float: + if time.monotonic() - slice_start < _EVENT_LOOP_SLICE_SECONDS: + return slice_start + _charge_inline_cpu() + await _yield_to_event_loop() + _restart_inline_cpu() + slice_start = time.monotonic() + if request_scoped: + _scan_slice_start.set(slice_start) + return slice_start + + class _SusPatternsRegexMixin(_SusPatternsRegistryMixin, _SusPatternsEnhancedMixin): async def _check_regex_pattern( self, @@ -388,8 +366,7 @@ async def _check_regex_pattern( scan_window_matcher = _PATTERN_SCAN_WINDOW_MATCHERS.get(pattern.pattern) if scan_window_matcher is not None: - with _record_inline_cpu(): - matches = scan_window_matcher(content, pattern) + matches = scan_window_matcher(content, pattern) threat = _first_accepted_regex_threat( iter(matches), pattern, @@ -470,7 +447,7 @@ async def _check_windowed_pattern( _timed_in_worker(lambda: list(finder(content)), worker_cpu) ) try: - matches = await _await_regex_future(future, timeout) + matches = await _suspended(_await_regex_future(future, timeout)) report_scan_success() _record_scan_time(sum(worker_cpu)) except concurrent.futures.TimeoutError: @@ -542,7 +519,7 @@ async def _check_pattern_with_timeout( _timed_in_worker(lambda: pattern.search(content, search_from), worker_cpu) ) try: - match = await _await_regex_future(future, timeout) + match = await _suspended(_await_regex_future(future, timeout)) report_scan_success() _record_scan_time(sum(worker_cpu)) return match, False @@ -604,68 +581,73 @@ async def _check_regex_patterns( performance_monitor = state.performance_monitor binary_prefix = build_binary_prefix(content) - slice_start = time.monotonic() - for pattern, contexts, category in all_patterns: - if _scan_time_budget_exhausted(): - budget_threat = _scan_time_budget_exhausted_threat( - _resolve_pattern_weight( - _SCAN_TIME_BUDGET_EXHAUSTED_PATTERN, "custom" + time_cap = _scan_time_cap.get() + slice_start, request_scoped = _pass_slice_start() + mark_token = _open_cpu_pass() + try: + for pattern, contexts, category in all_patterns: + if time_cap is not None and _scan_time_budget_exhausted(): + budget_threat = _scan_time_budget_exhausted_threat( + _resolve_pattern_weight( + _SCAN_TIME_BUDGET_EXHAUSTED_PATTERN, "custom" + ) ) + _warn_scan_time_cap_reached_once(ip_address) + threats.append(budget_threat) + matched_patterns.append(budget_threat["pattern"]) + timeouts.append(budget_threat["pattern"]) + break + if _pattern_should_be_skipped( + pattern, + contexts, + category, + raw_view_only=raw_view_only, + skip_filter=skip_filter, + normalized_context=normalized, + enabled_categories=enabled_categories, + url_decoded_view_only=url_decoded_view_only, + ): + continue + + pattern_start = time.monotonic() + + scan_content = content + threat, timeout_occurred = await self._check_regex_pattern( + pattern, + scan_content, + ip_address, + pattern_start, + category, + state=state, + context=validator_context, + binary_prefix=binary_prefix, ) - _warn_scan_time_cap_reached_once(ip_address) - threats.append(budget_threat) - matched_patterns.append(budget_threat["pattern"]) - timeouts.append(budget_threat["pattern"]) - break - if _pattern_should_be_skipped( - pattern, - contexts, - category, - raw_view_only=raw_view_only, - skip_filter=skip_filter, - normalized_context=normalized, - enabled_categories=enabled_categories, - url_decoded_view_only=url_decoded_view_only, - ): - continue - - pattern_start = time.monotonic() - - scan_content = content - threat, timeout_occurred = await self._check_regex_pattern( - pattern, - scan_content, - ip_address, - pattern_start, - category, - state=state, - context=validator_context, - binary_prefix=binary_prefix, - ) - pattern_end = time.monotonic() - if pattern_end - slice_start >= _EVENT_LOOP_SLICE_SECONDS: - await _yield_to_event_loop() - slice_start = time.monotonic() + slice_start = await _yield_once_slice_elapsed( + slice_start, request_scoped + ) - if timeout_occurred: - timeouts.append(pattern.pattern) - if not threat: - threat = _build_timeout_threat(pattern, category, pattern_start) + if timeout_occurred: + timeouts.append(pattern.pattern) + if not threat: + threat = _build_timeout_threat(pattern, category, pattern_start) + + if threat: + threats.append(threat) + matched_patterns.append(pattern.pattern) + + if performance_monitor: + await performance_monitor.record_metric( + pattern=pattern.pattern, + execution_time=time.monotonic() - pattern_start, + content_length=len(content), + matched=bool(threat), + timeout=timeout_occurred, + agent_handler=self.agent_handler, + correlation_id=correlation_id, + ) - if threat: - threats.append(threat) - matched_patterns.append(pattern.pattern) - - if performance_monitor: - await performance_monitor.record_metric( - pattern=pattern.pattern, - execution_time=time.monotonic() - pattern_start, - content_length=len(content), - matched=bool(threat), - timeout=timeout_occurred, - agent_handler=self.agent_handler, - correlation_id=correlation_id, - ) + finally: + _close_cpu_pass(mark_token) return threats, matched_patterns, timeouts diff --git a/guard_core/handlers/_suspatterns_scan_windows.py b/guard_core/handlers/_suspatterns_scan_windows.py new file mode 100644 index 00000000..b380b4c2 --- /dev/null +++ b/guard_core/handlers/_suspatterns_scan_windows.py @@ -0,0 +1,55 @@ +import re +from collections.abc import Iterator + +from guard_core.detection_engine.scan_window import bounded_finditer +from guard_core.handlers._suspatterns_xml_xxe import ( + _xml_internal_entity_finditer, + _xml_system_finditer, +) + +_SCAN_WINDOW_BOUND_SOURCES: dict[str, tuple[tuple[str, str], ...]] = { + r"]*>[^<]*<\/script\s*>": ((r""),), + ( + r"(?:<[A-Za-z/][^<>]*style\s*=\s{0,20}[\"']?[^<>\"']*" + r"(?:expression|behavior|url)\s*\([^)]*\))" + ): ((r"<[A-Za-z/][^<>]*style\s*=", r"\)"),), + r"(?:]*>[\s\S]*<\/object\s*>)": ((r""),), + r"(?:]*>[\s\S]*<\/embed\s*>)": ((r""),), + r"(?:]*>[\s\S]*<\/applet\s*>)": ((r""),), + r"\.\.;[^/\\]*[/\\]": ((r"\.\.;", r"[/\\]"),), + ( + r"=(?:https?|ftp):\/\/[^\s'\"<>]+\/[^\s'\"<>\/]*" + r"\.(?:phtml|php[3-5]?|phar|jsp|aspx?|pl|py|txt|inc)(?![a-zA-Z0-9])" + ): ( + ( + r"=(?:https?|ftp):\/\/", + r"\.(?:phtml|php\d*|phar|jsp|aspx?|pl|py|txt|inc)[a-zA-Z0-9]*", + ), + ), + r"]+SYSTEM[^>]+>": ((r""),), + r"(?:)": ((r""),), + r"\[]*\[[\s\S]*? Iterator[re.Match]: + if pattern.pattern == r"]+SYSTEM[^>]+>": + yield from _xml_system_finditer(content) + return + if pattern.pattern == r"\[]*\[[\s\S]*? None: _scan_time_spent.set(_scan_time_spent.get() + elapsed) -@contextmanager -def _record_inline_cpu() -> Iterator[None]: - start = _thread_cpu_seconds() +def _open_cpu_pass() -> contextvars.Token[float | None]: + return _scan_cpu_mark.set( + None if _scan_time_cap.get() is None else _thread_cpu_seconds() + ) + + +def _close_cpu_pass(token: contextvars.Token[float | None]) -> None: + _charge_inline_cpu() + _scan_cpu_mark.reset(token) + + +def _charge_inline_cpu() -> None: + mark = _scan_cpu_mark.get() + if mark is not None: + now = _thread_cpu_seconds() + _record_scan_time(now - mark) + _scan_cpu_mark.set(now) + + +def _restart_inline_cpu() -> None: + if _scan_cpu_mark.get() is not None: + _scan_cpu_mark.set(_thread_cpu_seconds()) + + +def _suspended(awaitable: _T) -> _T: + _charge_inline_cpu() try: - yield + return awaitable finally: - _record_scan_time(_thread_cpu_seconds() - start) + _restart_inline_cpu() def _timed_in_worker( @@ -233,53 +267,6 @@ def _warn_json_depth_cap_reached_once(client_ip: str) -> None: ) -_PATTERN_SOURCE_NON_TOKEN_RE = re.compile(r"[^a-z0-9_-]+") -_PATTERN_SOURCE_ESCAPE_RE = re.compile( - r"\\(?:[xX]([0-9A-Fa-f]{2})|u([0-9A-Fa-f]{4})|U([0-9A-Fa-f]{8})|([0-7]{1,3}))" -) - - -def _decode_pattern_source_escape(match: re.Match[str]) -> str: - hex_body = match.group(1) or match.group(2) or match.group(3) - code = int(hex_body, 16) if hex_body else int(match.group(4), 8) - return chr(code) if code <= 0x10FFFF else match.group(0) - - -def _pattern_source_names_a_sensitive_field( - pattern_source: str, sensitive_names: frozenset[str] -) -> bool: - decoded = _PATTERN_SOURCE_ESCAPE_RE.sub( - _decode_pattern_source_escape, pattern_source - ) - normalized = _PATTERN_SOURCE_NON_TOKEN_RE.sub("", decoded.lower()) - return any(name in normalized for name in sensitive_names) - - -def _redact_pattern_source(pattern_source: str) -> str: - from guard_core.sync._utils.request_logging import ( - _merge_sensitive_log_body_fields, - _merged_sensitive_names, - redact_blob_for_display, - ) - from guard_core.sync.handlers.suspatterns_handler import sus_patterns_handler - - sensitive_params = sus_patterns_handler._sensitive_params_union - sensitive_body_fields = sus_patterns_handler._sensitive_body_fields_union - sensitive_headers = sus_patterns_handler._sensitive_headers_union - - sensitive_names = _merged_sensitive_names( - sensitive_params, - _merge_sensitive_log_body_fields(sensitive_body_fields), - sensitive_headers, - ) - if _pattern_source_names_a_sensitive_field(pattern_source, sensitive_names): - return "[REDACTED]" - - return redact_blob_for_display( - pattern_source, sensitive_params, sensitive_body_fields, sensitive_headers - ) - - def _build_threat_message(threat: dict[str, Any]) -> str: if threat["type"] == "regex": return f"Value matched pattern '{_redact_pattern_source(threat['pattern'])}'" @@ -329,31 +316,36 @@ def _fallback_pattern_check( normalized_context = sus_patterns_handler._normalize_context(context) all_compiled = sus_patterns_handler.get_all_compiled_patterns() binary_prefix = build_binary_prefix(value) - for pattern, _contexts, category in all_compiled: - if _scan_time_budget_exhausted(): - return True, "Scan time budget exhausted (fallback)" - pattern_start = time.monotonic() - try: - threat, _timeout_occurred = sus_patterns_handler._check_regex_pattern( - pattern, - value, - client_ip, - pattern_start, - category, - context=normalized_context, - binary_prefix=binary_prefix, - ) - except RecursionError: - logger.warning( - "Fallback pattern search hit the regex engine's recursion " - "limit on pattern %r; skipping it and continuing the " - "fallback scan", - _redact_pattern_source(pattern.pattern), - ) - continue - if threat: - return True, "Value matched pattern (fallback)" - return False, "" + mark_token = _open_cpu_pass() + try: + for pattern, _contexts, category in all_compiled: + _charge_inline_cpu() + if _scan_time_budget_exhausted(): + return True, "Scan time budget exhausted (fallback)" + pattern_start = time.monotonic() + try: + threat, _ = sus_patterns_handler._check_regex_pattern( + pattern, + value, + client_ip, + pattern_start, + category, + context=normalized_context, + binary_prefix=binary_prefix, + ) + except RecursionError: + logger.warning( + "Fallback pattern search hit the regex engine's recursion " + "limit on pattern %r; skipping it and continuing the " + "fallback scan", + _redact_pattern_source(pattern.pattern), + ) + continue + if threat: + return True, "Value matched pattern (fallback)" + return False, "" + finally: + _close_cpu_pass(mark_token) def _resolve_json_redact_all( @@ -559,17 +551,18 @@ def _check_always_scan_header(value: str) -> tuple[bool, str, list[dict]]: def _compiler_finditer( compiler: Any, pattern: re.Pattern, category: str, content: str ) -> list[re.Match]: - inline_scan = category != "custom" + if category != "custom": + inline_finder = compiler.create_async_safe_finditer_matcher( + pattern, inline_safe=True + ) + inline_matches: list[re.Match] = inline_finder(content) + return inline_matches worker_cpu: list[float] = [] finder = compiler.create_async_safe_finditer_matcher( pattern, - inline_safe=inline_scan, + inline_safe=False, worker_wrapper=lambda work: _timed_in_worker(work, worker_cpu), ) - if inline_scan: - with _record_inline_cpu(): - matches: list[re.Match] = finder(content) - return matches - matches = finder(content) + matches: list[re.Match] = _suspended(finder(content)) _record_scan_time(sum(worker_cpu) if worker_cpu else compiler.default_timeout) return matches diff --git a/guard_core/sync/_utils/pattern_source_redaction.py b/guard_core/sync/_utils/pattern_source_redaction.py new file mode 100644 index 00000000..5890f02d --- /dev/null +++ b/guard_core/sync/_utils/pattern_source_redaction.py @@ -0,0 +1,47 @@ +import re + +_PATTERN_SOURCE_NON_TOKEN_RE = re.compile(r"[^a-z0-9_-]+") +_PATTERN_SOURCE_ESCAPE_RE = re.compile( + r"\\(?:[xX]([0-9A-Fa-f]{2})|u([0-9A-Fa-f]{4})|U([0-9A-Fa-f]{8})|([0-7]{1,3}))" +) + + +def _decode_pattern_source_escape(match: re.Match[str]) -> str: + hex_body = match.group(1) or match.group(2) or match.group(3) + code = int(hex_body, 16) if hex_body else int(match.group(4), 8) + return chr(code) if code <= 0x10FFFF else match.group(0) + + +def _pattern_source_names_a_sensitive_field( + pattern_source: str, sensitive_names: frozenset[str] +) -> bool: + decoded = _PATTERN_SOURCE_ESCAPE_RE.sub( + _decode_pattern_source_escape, pattern_source + ) + normalized = _PATTERN_SOURCE_NON_TOKEN_RE.sub("", decoded.lower()) + return any(name in normalized for name in sensitive_names) + + +def _redact_pattern_source(pattern_source: str) -> str: + from guard_core.sync._utils.request_logging import ( + _merge_sensitive_log_body_fields, + _merged_sensitive_names, + redact_blob_for_display, + ) + from guard_core.sync.handlers.suspatterns_handler import sus_patterns_handler + + sensitive_params = sus_patterns_handler._sensitive_params_union + sensitive_body_fields = sus_patterns_handler._sensitive_body_fields_union + sensitive_headers = sus_patterns_handler._sensitive_headers_union + + sensitive_names = _merged_sensitive_names( + sensitive_params, + _merge_sensitive_log_body_fields(sensitive_body_fields), + sensitive_headers, + ) + if _pattern_source_names_a_sensitive_field(pattern_source, sensitive_names): + return "[REDACTED]" + + return redact_blob_for_display( + pattern_source, sensitive_params, sensitive_body_fields, sensitive_headers + ) diff --git a/guard_core/sync/detection_engine/monitor.py b/guard_core/sync/detection_engine/monitor.py index eaa42988..4ce27646 100644 --- a/guard_core/sync/detection_engine/monitor.py +++ b/guard_core/sync/detection_engine/monitor.py @@ -182,6 +182,8 @@ def _notify_callbacks( agent_handler: Any, correlation_id: str | None, ) -> None: + if not self.anomaly_callbacks: + return safe_anomaly = self._sanitize_anomaly_data(anomaly) for callback in self.anomaly_callbacks: diff --git a/guard_core/sync/detection_engine/monitor_anomalies.py b/guard_core/sync/detection_engine/monitor_anomalies.py index 7ceee79b..7bb8630d 100644 --- a/guard_core/sync/detection_engine/monitor_anomalies.py +++ b/guard_core/sync/detection_engine/monitor_anomalies.py @@ -48,8 +48,9 @@ def detect_statistical_anomaly( if anomaly_threshold >= 0 and metric.execution_time <= avg_time: return None - variance = math.fsum((t - avg_time) ** 2 for t in recent_times) / (sample_count - 1) - std_time = variance**0.5 + std_time = math.dist(recent_times, [avg_time] * sample_count) / math.sqrt( + sample_count - 1 + ) if std_time <= 0: return None diff --git a/guard_core/sync/handlers/_suspatterns_regex.py b/guard_core/sync/handlers/_suspatterns_regex.py index 0936c2d5..50aa0743 100644 --- a/guard_core/sync/handlers/_suspatterns_regex.py +++ b/guard_core/sync/handlers/_suspatterns_regex.py @@ -8,12 +8,18 @@ from guard_core.sync._utils.detection_scan import ( _EVENT_LOOP_SLICE_SECONDS, _SCAN_TIME_BUDGET_EXHAUSTED_PATTERN, + _charge_inline_cpu, + _close_cpu_pass, _compiler_finditer, - _record_inline_cpu, + _open_cpu_pass, _record_scan_time, _redact_pattern_source, + _restart_inline_cpu, + _scan_slice_start, _scan_time_budget_exhausted, _scan_time_budget_exhausted_threat, + _scan_time_cap, + _suspended, _timed_in_worker, _warn_scan_time_cap_reached_once, ) @@ -28,7 +34,6 @@ report_scan_timeout, shared_regex_executor, ) -from guard_core.sync.detection_engine.scan_window import bounded_finditer from guard_core.sync.handlers._suspatterns_enhanced import _SusPatternsEnhancedMixin from guard_core.sync.handlers._suspatterns_ldap_ipv4 import ( _LEGACY_IPV4_HOST_RE, @@ -54,6 +59,10 @@ _pickle_global_candidate_is_injection, ) from guard_core.sync.handlers._suspatterns_registry import _SusPatternsRegistryMixin +from guard_core.sync.handlers._suspatterns_scan_windows import ( + _SCAN_WINDOW_PATTERNS, + _iter_scan_window_matches, +) from guard_core.sync.handlers._suspatterns_shell_sources import ( _BRACE_EXPANSION_COMMAND_RE, _GLOB_WILDCARD_ATOM_RE, @@ -98,8 +107,6 @@ from guard_core.sync.handlers._suspatterns_state import _DetectionState from guard_core.sync.handlers._suspatterns_xml_xxe import ( _XML_XXE_PUBLIC_EXTERNAL_DTD_COMPILED_RE, - _xml_internal_entity_finditer, - _xml_system_finditer, _xml_xxe_public_external_dtd_finditer, ) @@ -242,53 +249,6 @@ def _pattern_should_be_skipped( ), } -_SCAN_WINDOW_BOUND_SOURCES: dict[str, tuple[tuple[str, str], ...]] = { - r"]*>[^<]*<\/script\s*>": ((r""),), - ( - r"(?:<[A-Za-z/][^<>]*style\s*=\s{0,20}[\"']?[^<>\"']*" - r"(?:expression|behavior|url)\s*\([^)]*\))" - ): ((r"<[A-Za-z/][^<>]*style\s*=", r"\)"),), - r"(?:]*>[\s\S]*<\/object\s*>)": ((r""),), - r"(?:]*>[\s\S]*<\/embed\s*>)": ((r""),), - r"(?:]*>[\s\S]*<\/applet\s*>)": ((r""),), - r"\.\.;[^/\\]*[/\\]": ((r"\.\.;", r"[/\\]"),), - ( - r"=(?:https?|ftp):\/\/[^\s'\"<>]+\/[^\s'\"<>\/]*" - r"\.(?:phtml|php[3-5]?|phar|jsp|aspx?|pl|py|txt|inc)(?![a-zA-Z0-9])" - ): ( - ( - r"=(?:https?|ftp):\/\/", - r"\.(?:phtml|php\d*|phar|jsp|aspx?|pl|py|txt|inc)[a-zA-Z0-9]*", - ), - ), - r"]+SYSTEM[^>]+>": ((r""),), - r"(?:)": ((r""),), - r"\[]*\[[\s\S]*? Iterator[re.Match]: - if pattern.pattern == r"]+SYSTEM[^>]+>": - yield from _xml_system_finditer(content) - return - if pattern.pattern == r"\[]*\[[\s\S]*? str: return value.encode("utf-8", errors="surrogateescape").decode( @@ -348,16 +308,34 @@ def _first_accepted_regex_threat( context: str = "unknown", binary_prefix: list[int] | None = None, ) -> dict[str, Any] | None: - with _record_inline_cpu(): - for match in matches: - threat = _build_regex_threat( - pattern, match, category, pattern_start, context, binary_prefix - ) - if threat: - return threat + for match in matches: + threat = _build_regex_threat( + pattern, match, category, pattern_start, context, binary_prefix + ) + if threat: + return threat return None +def _pass_slice_start() -> tuple[float, bool]: + request_slice_start = _scan_slice_start.get() + if request_slice_start is None: + return time.monotonic(), False + return request_slice_start, True + + +def _yield_once_slice_elapsed(slice_start: float, request_scoped: bool) -> float: + if time.monotonic() - slice_start < _EVENT_LOOP_SLICE_SECONDS: + return slice_start + _charge_inline_cpu() + _yield_to_event_loop() + _restart_inline_cpu() + slice_start = time.monotonic() + if request_scoped: + _scan_slice_start.set(slice_start) + return slice_start + + class _SusPatternsRegexMixin(_SusPatternsRegistryMixin, _SusPatternsEnhancedMixin): def _check_regex_pattern( self, @@ -388,8 +366,7 @@ def _check_regex_pattern( scan_window_matcher = _PATTERN_SCAN_WINDOW_MATCHERS.get(pattern.pattern) if scan_window_matcher is not None: - with _record_inline_cpu(): - matches = scan_window_matcher(content, pattern) + matches = scan_window_matcher(content, pattern) threat = _first_accepted_regex_threat( iter(matches), pattern, @@ -470,7 +447,7 @@ def _check_windowed_pattern( _timed_in_worker(lambda: list(finder(content)), worker_cpu) ) try: - matches = _await_regex_future(future, timeout) + matches = _suspended(_await_regex_future(future, timeout)) report_scan_success() _record_scan_time(sum(worker_cpu)) except concurrent.futures.TimeoutError: @@ -542,7 +519,7 @@ def _check_pattern_with_timeout( _timed_in_worker(lambda: pattern.search(content, search_from), worker_cpu) ) try: - match = _await_regex_future(future, timeout) + match = _suspended(_await_regex_future(future, timeout)) report_scan_success() _record_scan_time(sum(worker_cpu)) return match, False @@ -604,68 +581,71 @@ def _check_regex_patterns( performance_monitor = state.performance_monitor binary_prefix = build_binary_prefix(content) - slice_start = time.monotonic() - for pattern, contexts, category in all_patterns: - if _scan_time_budget_exhausted(): - budget_threat = _scan_time_budget_exhausted_threat( - _resolve_pattern_weight( - _SCAN_TIME_BUDGET_EXHAUSTED_PATTERN, "custom" + time_cap = _scan_time_cap.get() + slice_start, request_scoped = _pass_slice_start() + mark_token = _open_cpu_pass() + try: + for pattern, contexts, category in all_patterns: + if time_cap is not None and _scan_time_budget_exhausted(): + budget_threat = _scan_time_budget_exhausted_threat( + _resolve_pattern_weight( + _SCAN_TIME_BUDGET_EXHAUSTED_PATTERN, "custom" + ) ) + _warn_scan_time_cap_reached_once(ip_address) + threats.append(budget_threat) + matched_patterns.append(budget_threat["pattern"]) + timeouts.append(budget_threat["pattern"]) + break + if _pattern_should_be_skipped( + pattern, + contexts, + category, + raw_view_only=raw_view_only, + skip_filter=skip_filter, + normalized_context=normalized, + enabled_categories=enabled_categories, + url_decoded_view_only=url_decoded_view_only, + ): + continue + + pattern_start = time.monotonic() + + scan_content = content + threat, timeout_occurred = self._check_regex_pattern( + pattern, + scan_content, + ip_address, + pattern_start, + category, + state=state, + context=validator_context, + binary_prefix=binary_prefix, ) - _warn_scan_time_cap_reached_once(ip_address) - threats.append(budget_threat) - matched_patterns.append(budget_threat["pattern"]) - timeouts.append(budget_threat["pattern"]) - break - if _pattern_should_be_skipped( - pattern, - contexts, - category, - raw_view_only=raw_view_only, - skip_filter=skip_filter, - normalized_context=normalized, - enabled_categories=enabled_categories, - url_decoded_view_only=url_decoded_view_only, - ): - continue - - pattern_start = time.monotonic() - - scan_content = content - threat, timeout_occurred = self._check_regex_pattern( - pattern, - scan_content, - ip_address, - pattern_start, - category, - state=state, - context=validator_context, - binary_prefix=binary_prefix, - ) - pattern_end = time.monotonic() - if pattern_end - slice_start >= _EVENT_LOOP_SLICE_SECONDS: - _yield_to_event_loop() - slice_start = time.monotonic() - - if timeout_occurred: - timeouts.append(pattern.pattern) - if not threat: - threat = _build_timeout_threat(pattern, category, pattern_start) + slice_start = _yield_once_slice_elapsed(slice_start, request_scoped) + + if timeout_occurred: + timeouts.append(pattern.pattern) + if not threat: + threat = _build_timeout_threat(pattern, category, pattern_start) + + if threat: + threats.append(threat) + matched_patterns.append(pattern.pattern) + + if performance_monitor: + performance_monitor.record_metric( + pattern=pattern.pattern, + execution_time=time.monotonic() - pattern_start, + content_length=len(content), + matched=bool(threat), + timeout=timeout_occurred, + agent_handler=self.agent_handler, + correlation_id=correlation_id, + ) - if threat: - threats.append(threat) - matched_patterns.append(pattern.pattern) - - if performance_monitor: - performance_monitor.record_metric( - pattern=pattern.pattern, - execution_time=time.monotonic() - pattern_start, - content_length=len(content), - matched=bool(threat), - timeout=timeout_occurred, - agent_handler=self.agent_handler, - correlation_id=correlation_id, - ) + finally: + _close_cpu_pass(mark_token) return threats, matched_patterns, timeouts diff --git a/guard_core/sync/handlers/_suspatterns_scan_windows.py b/guard_core/sync/handlers/_suspatterns_scan_windows.py new file mode 100644 index 00000000..349c7a53 --- /dev/null +++ b/guard_core/sync/handlers/_suspatterns_scan_windows.py @@ -0,0 +1,55 @@ +import re +from collections.abc import Iterator + +from guard_core.sync.detection_engine.scan_window import bounded_finditer +from guard_core.sync.handlers._suspatterns_xml_xxe import ( + _xml_internal_entity_finditer, + _xml_system_finditer, +) + +_SCAN_WINDOW_BOUND_SOURCES: dict[str, tuple[tuple[str, str], ...]] = { + r"]*>[^<]*<\/script\s*>": ((r""),), + ( + r"(?:<[A-Za-z/][^<>]*style\s*=\s{0,20}[\"']?[^<>\"']*" + r"(?:expression|behavior|url)\s*\([^)]*\))" + ): ((r"<[A-Za-z/][^<>]*style\s*=", r"\)"),), + r"(?:]*>[\s\S]*<\/object\s*>)": ((r""),), + r"(?:]*>[\s\S]*<\/embed\s*>)": ((r""),), + r"(?:]*>[\s\S]*<\/applet\s*>)": ((r""),), + r"\.\.;[^/\\]*[/\\]": ((r"\.\.;", r"[/\\]"),), + ( + r"=(?:https?|ftp):\/\/[^\s'\"<>]+\/[^\s'\"<>\/]*" + r"\.(?:phtml|php[3-5]?|phar|jsp|aspx?|pl|py|txt|inc)(?![a-zA-Z0-9])" + ): ( + ( + r"=(?:https?|ftp):\/\/", + r"\.(?:phtml|php\d*|phar|jsp|aspx?|pl|py|txt|inc)[a-zA-Z0-9]*", + ), + ), + r"]+SYSTEM[^>]+>": ((r""),), + r"(?:)": ((r""),), + r"\[]*\[[\s\S]*? Iterator[re.Match]: + if pattern.pattern == r"]+SYSTEM[^>]+>": + yield from _xml_system_finditer(content) + return + if pattern.pattern == r"\[]*\[[\s\S]*? None: assert ran_at assert ran_at[0] < len(all_patterns) assert ran_at[0] <= 2 + + +def _fake_time(now: list[float]) -> Any: + return SimpleNamespace(monotonic=lambda: now[0], thread_time=time.thread_time) + + +async def test_slice_clock_spans_values_within_a_request() -> None: + manager = SusPatternsManager() + manager.configure(SecurityConfig()) + now = [0.0] + checks = {"count": 0} + yields = {"count": 0} + real_check = manager._check_regex_pattern + real_yield = regex_module._yield_to_event_loop + + async def counting_check(*args: Any, **kwargs: Any) -> Any: + checks["count"] += 1 + now[0] += step[0] + return await real_check(*args, **kwargs) + + async def counting_yield() -> None: + yields["count"] += 1 + await real_yield() + + step = [0.0] + with ( + patch.object(manager, "_check_regex_pattern", counting_check), + patch.object(regex_module, "_yield_to_event_loop", counting_yield), + patch.object(regex_module, "time", _fake_time(now)), + patch.object(detection_scan, "time", _fake_time(now)), + ): + await manager.detect("hello world", "127.0.0.1", "query_param:q") + per_value = checks["count"] + step[0] = 0.0015 / per_value + with _scan_value_budget(512): + await manager.detect("hello world", "127.0.0.1", "query_param:q") + assert yields["count"] == 0 + await manager.detect("hello world", "127.0.0.1", "query_param:q") + assert yields["count"] >= 1 + + +async def test_suspended_time_is_not_charged_to_the_request() -> None: + manager = SusPatternsManager() + manager.configure(SecurityConfig()) + + async def burn_cpu() -> None: + end = time.thread_time() + 0.3 + while time.thread_time() < end: + pass + + burner = asyncio.create_task(burn_cpu()) + with _scan_value_budget(512, max_scan_seconds=0.15): + result = await manager.detect("hello world", "127.0.0.1", "query_param:q") + await burner + + assert all(t["pattern"] != "scan_time_budget_exhausted" for t in result["threats"]) diff --git a/tests/test_scan_time_budget.py b/tests/test_scan_time_budget.py index a331662f..696d6c4b 100644 --- a/tests/test_scan_time_budget.py +++ b/tests/test_scan_time_budget.py @@ -1,4 +1,5 @@ import asyncio +import threading from collections.abc import AsyncGenerator, Iterator from typing import Any from unittest.mock import patch @@ -11,7 +12,10 @@ _scan_time_budget_exhausted, _scan_value_budget, ) -from guard_core.handlers.suspatterns_handler import SusPatternsManager +from guard_core.handlers.suspatterns_handler import ( + SusPatternsManager, + sus_patterns_handler, +) from guard_core.models import SecurityConfig from guard_core.utils import detect_penetration_attempt from tests.conftest import MockGuardRequest @@ -153,6 +157,24 @@ async def test_fallback_check_honours_budget() -> None: assert "budget" in trigger +async def test_fallback_check_charges_its_inline_scan_cpu( + fake_regex_clock: _FakeClock, +) -> None: + async def inline_only(*args: Any, **kwargs: Any) -> tuple[None, bool]: + return None, False + + with ( + patch.object(sus_patterns_handler, "_check_regex_pattern", inline_only), + _scan_value_budget(512, max_scan_seconds=2.5), + ): + detected, trigger = await detection_scan._fallback_pattern_check( + "hello", "127.0.0.1", "query_param:q" + ) + + assert detected is True + assert "budget" in trigger + + async def test_trigger_names_the_budget_when_it_tips_the_request( fake_regex_clock: _FakeClock, ) -> None: @@ -217,3 +239,49 @@ async def test_worker_scan_cpu_is_recorded_from_the_worker_thread( assert remaining is not None assert remaining < 1000.0 + + +async def test_inline_cpu_is_accounted_per_pass_not_per_pattern() -> None: + manager = SusPatternsManager() + manager.configure(SecurityConfig()) + calls = {"clock": 0, "patterns": 0} + real = manager._check_regex_pattern + + scan_thread = threading.get_ident() + + def counting_clock() -> float: + if threading.get_ident() == scan_thread: + calls["clock"] += 1 + return 0.0 + + async def counting(*args: Any, **kwargs: Any) -> Any: + calls["patterns"] += 1 + return await real(*args, **kwargs) + + with ( + patch.object(detection_scan, "_thread_cpu_seconds", counting_clock), + patch.object(manager, "_check_regex_pattern", counting), + _scan_value_budget(512, max_scan_seconds=1000.0), + ): + await manager.detect("hello world", "127.0.0.1", "query_param:q") + + assert calls["patterns"] > 50 + assert 0 < calls["clock"] < calls["patterns"] * 0.2 + + +async def test_no_cpu_clock_reads_outside_a_request_scope() -> None: + manager = SusPatternsManager() + manager.configure(SecurityConfig()) + calls = {"clock": 0} + + scan_thread = threading.get_ident() + + def counting_clock() -> float: + if threading.get_ident() == scan_thread: + calls["clock"] += 1 + return 0.0 + + with patch.object(detection_scan, "_thread_cpu_seconds", counting_clock): + await manager.detect("hello world", "127.0.0.1", "query_param:q") + + assert calls["clock"] == 0 diff --git a/tests/test_sus_patterns/test_monitor.py b/tests/test_sus_patterns/test_monitor.py index fd774e1a..772e54e2 100644 --- a/tests/test_sus_patterns/test_monitor.py +++ b/tests/test_sus_patterns/test_monitor.py @@ -907,6 +907,15 @@ def bad_callback(anomaly: dict[str, Any]) -> None: assert len(captured) == 1 +async def test_notify_callbacks_skips_sanitizing_without_callbacks() -> None: + monitor = PerformanceMonitor() + with patch.object(monitor, "_sanitize_anomaly_data") as sanitize: + await monitor._notify_callbacks( + {"pattern": "x", "type": "t"}, agent_handler=None, correlation_id=None + ) + sanitize.assert_not_called() + + async def test_get_slow_patterns_skips_missing_report() -> None: monitor = PerformanceMonitor() stats = PatternStats(pattern="gone") diff --git a/tests/test_sync/test_scan_time_budget.py b/tests/test_sync/test_scan_time_budget.py index 4d0a071e..1abb4d62 100644 --- a/tests/test_sync/test_scan_time_budget.py +++ b/tests/test_sync/test_scan_time_budget.py @@ -1,3 +1,4 @@ +import threading import time from collections.abc import Generator, Iterator from typing import Any @@ -12,7 +13,10 @@ _scan_time_budget_exhausted, _scan_value_budget, ) -from guard_core.sync.handlers.suspatterns_handler import SusPatternsManager +from guard_core.sync.handlers.suspatterns_handler import ( + SusPatternsManager, + sus_patterns_handler, +) from guard_core.sync.utils import detect_penetration_attempt from tests.test_sync.conftest import SyncMockGuardRequest @@ -155,6 +159,24 @@ def test_fallback_check_honours_budget() -> None: assert "budget" in trigger +def test_fallback_check_charges_its_inline_scan_cpu( + fake_regex_clock: _FakeClock, +) -> None: + def inline_only(*args: Any, **kwargs: Any) -> tuple[None, bool]: + return None, False + + with ( + patch.object(sus_patterns_handler, "_check_regex_pattern", inline_only), + _scan_value_budget(512, max_scan_seconds=2.5), + ): + detected, trigger = detection_scan._fallback_pattern_check( + "hello", "127.0.0.1", "query_param:q" + ) + + assert detected is True + assert "budget" in trigger + + def test_trigger_names_the_budget_when_it_tips_the_request( fake_regex_clock: _FakeClock, ) -> None: @@ -219,3 +241,49 @@ def test_worker_scan_cpu_is_recorded_from_the_worker_thread( assert remaining is not None assert remaining < 1000.0 + + +def test_inline_cpu_is_accounted_per_pass_not_per_pattern() -> None: + manager = SusPatternsManager() + manager.configure(SecurityConfig()) + calls = {"clock": 0, "patterns": 0} + real = manager._check_regex_pattern + + scan_thread = threading.get_ident() + + def counting_clock() -> float: + if threading.get_ident() == scan_thread: + calls["clock"] += 1 + return 0.0 + + def counting(*args: Any, **kwargs: Any) -> Any: + calls["patterns"] += 1 + return real(*args, **kwargs) + + with ( + patch.object(detection_scan, "_thread_cpu_seconds", counting_clock), + patch.object(manager, "_check_regex_pattern", counting), + _scan_value_budget(512, max_scan_seconds=1000.0), + ): + manager.detect("hello world", "127.0.0.1", "query_param:q") + + assert calls["patterns"] > 50 + assert 0 < calls["clock"] < calls["patterns"] * 0.2 + + +def test_no_cpu_clock_reads_outside_a_request_scope() -> None: + manager = SusPatternsManager() + manager.configure(SecurityConfig()) + calls = {"clock": 0} + + scan_thread = threading.get_ident() + + def counting_clock() -> float: + if threading.get_ident() == scan_thread: + calls["clock"] += 1 + return 0.0 + + with patch.object(detection_scan, "_thread_cpu_seconds", counting_clock): + manager.detect("hello world", "127.0.0.1", "query_param:q") + + assert calls["clock"] == 0 diff --git a/tests/test_sync/test_sus_patterns/test_monitor.py b/tests/test_sync/test_sus_patterns/test_monitor.py index 6d4865f0..67789183 100644 --- a/tests/test_sync/test_sus_patterns/test_monitor.py +++ b/tests/test_sync/test_sus_patterns/test_monitor.py @@ -862,6 +862,15 @@ def bad_callback(anomaly: dict[str, Any]) -> None: assert len(captured) == 1 +def test_notify_callbacks_skips_sanitizing_without_callbacks() -> None: + monitor = PerformanceMonitor() + with patch.object(monitor, "_sanitize_anomaly_data") as sanitize: + monitor._notify_callbacks( + {"pattern": "x", "type": "t"}, agent_handler=None, correlation_id=None + ) + sanitize.assert_not_called() + + def test_get_slow_patterns_skips_missing_report() -> None: monitor = PerformanceMonitor() stats = PatternStats(pattern="gone")