From 24b2f0b53223a051e44e700127b674c738c0cd7d Mon Sep 17 00:00:00 2001 From: yashrajbasav Date: Mon, 5 Oct 2026 10:55:17 +0530 Subject: [PATCH 01/34] fix(security): bound static regex matching with runtime deadlines Signed-off-by: yashrajbasav --- .../static_patterns_agent_snooping.py | 6 +- .../static_patterns_data_exfiltration.py | 10 +- .../static_patterns_excessive_agency.py | 6 +- .../static_patterns_memory_poisoning.py | 2 +- .../static_patterns_prompt_injection.py | 2 +- .../analyzers/static_patterns_rogue_agent.py | 4 +- .../analyzers/static_patterns_supply_chain.py | 16 +- .../analyzers/static_patterns_tool_misuse.py | 10 +- .../nodes/analyzers/static_runner.py | 193 +++++++++++++++++- tests/nodes/analyzers/test_static_patterns.py | 90 ++++++++ 10 files changed, 313 insertions(+), 26 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_patterns_agent_snooping.py b/src/skillspector/nodes/analyzers/static_patterns_agent_snooping.py index 2bbd7b968..5747823c3 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_agent_snooping.py +++ b/src/skillspector/nodes/analyzers/static_patterns_agent_snooping.py @@ -153,7 +153,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in AS1_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = get_line_number(content, match.start()) @@ -175,7 +175,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in AS2_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = get_line_number(content, match.start()) @@ -197,7 +197,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in AS3_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): full_match = match.group(0) diff --git a/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py b/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py index ccb5edd4c..3c1b74018 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py +++ b/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py @@ -302,7 +302,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in E1_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = get_line_number(content, match.start()) @@ -337,7 +337,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in E2_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = get_line_number(content, match.start()) @@ -358,7 +358,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in E3_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = get_line_number(content, match.start()) @@ -395,7 +395,9 @@ def ctx(start: int) -> str: ) # E5: cloud-storage exfiltration. Example filtering is delegated to the runner. for pattern, confidence in E5_PATTERNS: - for match in re.finditer(pattern, content, re.IGNORECASE | re.MULTILINE): + for match in static_runner.iter_pattern_matches( + pattern, content, re.IGNORECASE | re.MULTILINE + ): line_num = get_line_number(content, match.start()) findings.append( AnalyzerFinding( diff --git a/src/skillspector/nodes/analyzers/static_patterns_excessive_agency.py b/src/skillspector/nodes/analyzers/static_patterns_excessive_agency.py index 13b83511e..15b7f0bc2 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_excessive_agency.py +++ b/src/skillspector/nodes/analyzers/static_patterns_excessive_agency.py @@ -386,7 +386,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in EA1_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = get_line_number(content, match.start()) @@ -407,7 +407,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in EA2_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = get_line_number(content, match.start()) @@ -447,7 +447,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in EA4_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = get_line_number(content, match.start()) diff --git a/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py b/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py index e160b167a..46dccb40d 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py +++ b/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py @@ -544,7 +544,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in MP2_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): span = match.group(0) diff --git a/src/skillspector/nodes/analyzers/static_patterns_prompt_injection.py b/src/skillspector/nodes/analyzers/static_patterns_prompt_injection.py index 4bf1ea076..23bf83dcf 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_prompt_injection.py +++ b/src/skillspector/nodes/analyzers/static_patterns_prompt_injection.py @@ -261,7 +261,7 @@ def _p2_pattern_matches( check_runtime() compiled = re.compile(pattern, re.IGNORECASE | re.DOTALL) if pattern not in _SINGLE_CHARACTER_P2_PATTERNS: - for match in compiled.finditer(content): + for match in static_runner.iter_pattern_matches(compiled, content): if check_runtime is not None: check_runtime() yield match diff --git a/src/skillspector/nodes/analyzers/static_patterns_rogue_agent.py b/src/skillspector/nodes/analyzers/static_patterns_rogue_agent.py index 1ce7a0f4f..535111d7a 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_rogue_agent.py +++ b/src/skillspector/nodes/analyzers/static_patterns_rogue_agent.py @@ -241,7 +241,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in RA1_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = bisect_right(line_starts, match.start()) @@ -293,7 +293,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in RA2_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = bisect_right(line_starts, match.start()) diff --git a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py index 10378c728..3497fcecd 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py +++ b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py @@ -1885,7 +1885,9 @@ def _iter_sc2_shell_matches( covered = 0 for start, end in command_ranges: if end is None: - yield from compiled.finditer(content, max(start, covered)) + yield from static_runner.iter_pattern_matches( + compiled, content, start=max(start, covered) + ) return if start < covered: continue @@ -1927,7 +1929,7 @@ def line_number(start: int) -> int: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in SC1_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.MULTILINE): line_num = line_number(match.start()) @@ -1952,7 +1954,9 @@ def line_number(start: int) -> int: pattern, content, re.IGNORECASE | re.MULTILINE ) else: - matches = re.finditer(pattern, content, re.IGNORECASE | re.MULTILINE) + matches = static_runner.iter_pattern_matches( + pattern, content, re.IGNORECASE | re.MULTILINE + ) for match in matches: line_num = line_number(match.start()) mt = match.group(0) @@ -2042,7 +2046,7 @@ def line_number(start: int) -> int: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in SC3_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = line_number(match.start()) @@ -2061,7 +2065,9 @@ def line_number(start: int) -> int: ) # SC7: untrusted container image. Example filtering is delegated to the runner. for pattern, confidence in SC7_PATTERNS: - for match in re.finditer(pattern, content, re.IGNORECASE | re.MULTILINE): + for match in static_runner.iter_pattern_matches( + pattern, content, re.IGNORECASE | re.MULTILINE + ): line_num = line_number(match.start()) findings.append( AnalyzerFinding( diff --git a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py index 66cee7e10..9c8cfc891 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py +++ b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py @@ -3127,7 +3127,7 @@ def _tm1_candidates( matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in TM1_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): yield match.start(), match.end(), match.group(0), confidence @@ -3661,7 +3661,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in TM2_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = get_line_number(content, match.start()) @@ -3691,7 +3691,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in TM3_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = get_line_number(content, match.start()) @@ -3710,7 +3710,9 @@ def ctx(start: int) -> str: ) # TM4: privileged K8s workload. Example filtering is delegated to the runner. for pattern, confidence in TM4_PATTERNS: - for match in re.finditer(pattern, content, re.IGNORECASE | re.MULTILINE): + for match in static_runner.iter_pattern_matches( + pattern, content, re.IGNORECASE | re.MULTILINE + ): line_num = get_line_number(content, match.start()) findings.append( AnalyzerFinding( diff --git a/src/skillspector/nodes/analyzers/static_runner.py b/src/skillspector/nodes/analyzers/static_runner.py index bb8f2b100..13977aca0 100644 --- a/src/skillspector/nodes/analyzers/static_runner.py +++ b/src/skillspector/nodes/analyzers/static_runner.py @@ -17,11 +17,13 @@ from __future__ import annotations +import functools import inspect import json import math import os import re +import sys import time import unicodedata from array import array @@ -33,6 +35,8 @@ from itertools import chain from typing import cast +import regex # type: ignore[import-untyped] + from skillspector.artifacts import ( ContentKind, SecurityTextView, @@ -294,17 +298,197 @@ def _paragraph_ranges(content: str) -> tuple[tuple[int, int], ...]: return result +_STATIC_PATTERN_SECONDS = 0.25 +_ACTIVE_FINDING_BUDGET: ContextVar[_FindingBudget | None] = ContextVar( + "static_pattern_finding_budget", default=None +) + + +@functools.lru_cache(maxsize=2) +def _python_category_ranges(ascii_only: bool) -> dict[str, str]: + """Keep the runtime's Python word/space/digit alphabet, not regex's Unicode version.""" + categories: dict[str, list[tuple[int, int]]] = {key: [] for key in "wWsSdD"} + budget = _ACTIVE_FINDING_BUDGET.get() + for value in range(sys.maxunicode + 1): + if budget is not None and value % 4096 == 0: + budget.check_runtime() + character = chr(value) + word = (character.isalnum() or character == "_") and (not ascii_only or value < 128) + space = character.isspace() and (not ascii_only or character in " \t\n\r\f\v") + digit = character.isdecimal() and (not ascii_only or value < 128) + for key in ("w" if word else "W", "s" if space else "S", "d" if digit else "D"): + ranges = categories[key] + if ranges and ranges[-1][1] == value - 1: + ranges[-1] = (ranges[-1][0], value) + else: + ranges.append((value, value)) + return { + key: "".join( + rf"\U{start:08x}" if start == end else rf"\U{start:08x}-\U{end:08x}" + for start, end in ranges + ) + for key, ranges in categories.items() + } + + +@functools.lru_cache(maxsize=512) +def _timed_pattern(source: str, flags: int) -> regex.Pattern[str]: + """Translate the static-rule grammar while retaining original match offsets. + + Scoped flags and verbose rules need their own grammar support. Reject them + as unavailable coverage rather than silently changing detection semantics. + """ + if flags & re.VERBOSE: + raise _StaticResourceLimitError(LedgerReason.RULES_UNAVAILABLE, {}) + timed_flags = regex.VERSION0 + for name in ("ASCII", "IGNORECASE", "MULTILINE", "DOTALL", "VERBOSE"): + if flags & getattr(re, name): + timed_flags |= getattr(regex, name) + + def literal_class(token: str) -> str: + if flags & re.IGNORECASE and not flags & re.ASCII: + for alias in ("ı", "İ"): + if bool(re.fullmatch(token, alias, flags)) != bool( + regex.fullmatch(token, alias, timed_flags) + ): + insert = 2 if token.startswith("[^") else 1 + token = token[:insert] + alias + token[insert:] + return token + + categories = _python_category_ranges(bool(flags & re.ASCII)) + word = "(?-i:[" + categories["w"] + "])" + parts: list[str] = [] + cursor = 0 + while cursor < len(source): + # Group names are syntax, never case-insensitive literals. + if source.startswith(("(?P<", "(?P="), cursor): + end = source.index(">" if source.startswith("(?P<", cursor) else ")", cursor) + 1 + parts.append(source[cursor:end]) + cursor = end + continue + flag_group = re.match(r"\(\?[aiLmsux-]+([:)])", source[cursor:]) + if flag_group is not None: + if flag_group.group(1) == ":": + raise _StaticResourceLimitError(LedgerReason.RULES_UNAVAILABLE, {}) + parts.append(flag_group.group(0)) + cursor += len(flag_group.group(0)) + continue + char = source[cursor] + if char == "\\" and cursor + 1 < len(source): + code = source[cursor + 1] + if code in "xuU": + end = cursor + {"x": 4, "u": 6, "U": 10}[code] + value = chr(int(source[cursor + 2 : end], 16)) + parts.append( + "[iIİı]" + if value in "iI" and flags & re.IGNORECASE and not flags & re.ASCII + else source[cursor:end] + ) + cursor = end + continue + if code in "NB": + raise _StaticResourceLimitError(LedgerReason.RULES_UNAVAILABLE, {}) + if code in categories: + parts.append("(?-i:[" + categories[code] + "])") + elif code == "b": + parts.append(rf"(?:(? Iterator[re.Match[str]]: + """Interrupt a single pattern search and retain the runner's partial evidence.""" + budget = _ACTIVE_FINDING_BUDGET.get() + if budget is not None: + budget.check_runtime() + original = re.compile(pattern, flags) + compiled = _timed_pattern(original.pattern, original.flags) + timeout = _STATIC_PATTERN_SECONDS + if budget is not None: + budget.check_runtime() + timeout = min(timeout, max(0.0, budget.deadline - budget.clock())) + started_at = time.monotonic() + try: + for match in compiled.finditer( + content, + start, + len(content) if end is None else end, + timeout=timeout, + concurrent=False, + ): + if budget is not None: + budget.check_runtime() + yield cast(re.Match[str], match) + except TimeoutError as exc: + raise _StaticResourceLimitError( + LedgerReason.RUNTIME_LIMIT, + { + "observed_seconds": max(0.0, time.monotonic() - started_at), + "limit_seconds": timeout, + }, + ) from exc + + def iter_paragraph_matches( pattern: str | re.Pattern[str], content: str, flags: int = 0 ) -> Iterator[re.Match[str]]: """Match prose within paragraphs; executable and structured rules use finditer.""" - regex = re.compile(pattern, flags) ranges = _paragraph_ranges(content) if not ranges: - yield from regex.finditer(content) + yield from iter_pattern_matches(pattern, content, flags) return for start, end in ranges: - yield from regex.finditer(content, start, end) + yield from iter_pattern_matches(pattern, content, flags, start=start, end=end) def security_view_match_is_literal(content: str, start: int, end: int) -> bool: @@ -895,6 +1079,7 @@ def _scan_path( module_finding_start = len(findings) occurrence_columns = _OccurrenceColumnResolver(content, line_starts) finding_budget.begin_module() + budget_token = _ACTIVE_FINDING_BUDGET.set(finding_budget) try: with observe_analyzer_findings(finding_budget.observe_creation): prepared = (prepared_analyses or {}).get(id(module)) @@ -951,6 +1136,8 @@ def _scan_path( if converted is not None: findings.append(converted) return findings, exc + finally: + _ACTIVE_FINDING_BUDGET.reset(budget_token) return findings, None diff --git a/tests/nodes/analyzers/test_static_patterns.py b/tests/nodes/analyzers/test_static_patterns.py index 35a7d12e5..2972cd9ba 100644 --- a/tests/nodes/analyzers/test_static_patterns.py +++ b/tests/nodes/analyzers/test_static_patterns.py @@ -17,6 +17,7 @@ from __future__ import annotations +import re from pathlib import Path from time import perf_counter from unittest.mock import MagicMock @@ -2177,3 +2178,92 @@ def test_license_ledger_records_ea3_for_non_boilerplate_content(self) -> None: assert result["inspection_ledger"][0]["outcome"] == "completed" assert result["inspection_ledger"][0]["path"] == path assert result["inspection_ledger"][0]["emitted_finding_ids"] == [f.finding_id for f in ea3] + + +@pytest.mark.parametrize( + ("pattern", "text", "flags"), + [ + (r"ignore\s+previous", "ıgnore\x1cprevious", re.IGNORECASE), + (r"(?i)ignore", "ıgnore", 0), + (r"\u0069gnore", "ıgnore", re.IGNORECASE), + (r"\x69gnore", "ıgnore", re.IGNORECASE), + (r"[A-Z]+", "İıſK", re.IGNORECASE), + (r"[^i]+", "ıİiIabc", re.IGNORECASE), + (r"\bword\b", "word\u0301", re.IGNORECASE), + (r"\bignore\b", "ignore\ua7cb", re.IGNORECASE), + (r"[\w.-]+", "word\ua7cb", re.IGNORECASE), + (r"[\W.-]+", "word\ua7cb", re.IGNORECASE), + (r"[^\w.-]+", "word\ua7cb", re.IGNORECASE), + (r"\d+", "1\U00010d40", re.IGNORECASE), + (r"[\d.]+", "1.\U00010d40", re.IGNORECASE), + (r"[^\d.]+", "1.\U00010d40", re.IGNORECASE), + (r"[^\s/]+", "path file/name", re.IGNORECASE), + (r"[^\n]*", "one\ntwo", re.IGNORECASE), + (r"[\w.-]+", "café\u0301.next", re.IGNORECASE), + (r"\w+\s+\S+", "café\x1eexample", re.IGNORECASE), + (r"(?P[A-Z]+)", "Skill", re.IGNORECASE), + (r"[A-Z]+\s+\w+", "ASCII data café", re.IGNORECASE | re.ASCII), + (r"one.*two", "one\ntwo", re.DOTALL), + ], +) +def test_timed_patterns_preserve_python_alphabet_and_evidence(pattern, text, flags): + expected = [ + (match.span(), match.group(0), match.groupdict()) + for match in re.finditer(pattern, text, flags) + ] + actual = [ + (match.span(), match.group(0), match.groupdict()) + for match in static_runner.iter_pattern_matches(pattern, text, flags) + ] + assert actual == expected + + +def test_static_regex_deadline_retains_findings_and_incomplete_ledger(monkeypatch): + original = static_runner._timed_pattern + + def compile_with_small_comment_deadline(source, flags): + if source.startswith("", "