diff --git a/docs/ANALYSIS_RESOURCE_BOUNDS.md b/docs/ANALYSIS_RESOURCE_BOUNDS.md index 812d19581..2344b4971 100644 --- a/docs/ANALYSIS_RESOURCE_BOUNDS.md +++ b/docs/ANALYSIS_RESOURCE_BOUNDS.md @@ -255,6 +255,7 @@ removes the rejected partial checkout. | Static findings | 10,000 | One artifact | | Static findings | 10,000 | One analyzer | | Static-analysis time | 300 seconds | One artifact, within the workflow deadline | +| Timed static-pattern matching CPU | 0.25 seconds | One rule per content window, shared across paragraphs, within the artifact and workflow deadlines | | YARA rule-directory entries | 10,000 | Built-in and optional directories combined | | YARA rule files | 1,024 | One rule load | | YARA rule source bytes | 1 MiB | One rule file | @@ -333,6 +334,47 @@ default. Set `SKILLSPECTOR_MAX_STATIC_ANALYSIS_SECONDS_PER_ARTIFACT` to a positi finite number of seconds to change that allowance. Invalid, zero, negative, infinite, or NaN values log a warning and retain the 300-second default. +The shared matcher has a fixed 0.25-second matching CPU limit per rule and content window. +Paragraph-matched rules share this allowance across all their paragraphs. +Ranges of eight characters or fewer use batched native searches. Longer ranges +retain the interruptible engine; timer setup can make paragraph-dense windows +roughly two to three times slower than untimed matching. This keeps their +backtracking bounded rather than extending the native exception to larger inputs. +Raising the artifact allowance does not raise this limit. It covers routed rules +in prompt injection, tool misuse, data exfiltration, supply chain, agent snooping, +excessive agency, memory poisoning, and rogue agents. Through paragraph matching, +it also covers prose rules in harmful content, system prompt leakage, output +handling, anti-refusal, privilege escalation, and SSRF. Output-handling and +privilege-escalation code patterns also use the timed matcher. Other code rules (SSRF endpoint/request patterns, +harmful-content substance names, and deserialization DS1–DS4) and helper-specific +searches remain outside it. The anti-refusal schema-field helper uses a linear +whitespace prefix so its benign-context check cannot backtrack quadratically. + +Audited command/option patterns use a linear, command-segment search that preserves +unbounded command spans and the original greedy matches. The SC3 shell-decoder +rules keep their native matcher: their literals, stages and options already have +explicit length/count bounds and atomic or possessive ambiguous groups. Memory poisoning's +2–20-character repetition rule uses Python's native engine: failed candidates +examine at most 20 copies, and successful candidates consume the repeated run. +The variable shell-flag rule also retains Python’s native backreference semantics. +It checks assignment headers separately, considers every eligible call in the +four-line assignment gap, and stops its native search 4,096 characters after +the last eligible call head. This is a conservative bound: an unterminated tail +reports `static_parse_limit` even if no later `shell=` exists. In-window matches +are retained. Both variable-name passes share one accumulated matching allowance, +and each analyzer window computes them once. Whole-artifact reconciliation gets +0.25 seconds per 256,000 characters. A variable-shell limit is reported after the +other lexical tool-misuse rules run, so it cannot hide their findings. +The repetition rule is bounded by the 256,000-character inspection window and the +artifact deadline, rather than the slower engine's per-search timer. + +Matching CPU accounting excludes caller work and other threads. The interruptible +engine itself measures process CPU, however, so native work in competing threads +can interrupt a search early. Retrying discards progress and can still exhaust the +matching allowance under sustained contention. If a search expires, its ledger +event keeps the observed time and limit, and that analyzer records partial coverage +for the artifact while retaining findings already produced. + ## Configuring the dependency-source deadline Dependency-source redirection analysis allows up to 5 seconds by default. Set diff --git a/src/skillspector/nodes/analyzers/static_patterns_agent_snooping.py b/src/skillspector/nodes/analyzers/static_patterns_agent_snooping.py index 23e47abdb..1182e7f6d 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_agent_snooping.py +++ b/src/skillspector/nodes/analyzers/static_patterns_agent_snooping.py @@ -154,7 +154,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()) @@ -176,7 +176,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()) @@ -196,7 +196,9 @@ def ctx(start: int) -> str: for pattern, confidence in AS3_PATTERNS: is_prose = (pattern, confidence) in AS3_PROSE_PATTERNS - matches = static_runner.iter_paragraph_matches if is_prose else re.finditer + matches = ( + static_runner.iter_paragraph_matches if is_prose else static_runner.iter_pattern_matches + ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): if is_prose and is_directly_prohibited(content, match.start(), match.end()): continue diff --git a/src/skillspector/nodes/analyzers/static_patterns_anti_refusal.py b/src/skillspector/nodes/analyzers/static_patterns_anti_refusal.py index a05442c39..62329a1e2 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_anti_refusal.py +++ b/src/skillspector/nodes/analyzers/static_patterns_anti_refusal.py @@ -40,7 +40,7 @@ from skillspector.state import AnalyzerNodeResponse, SkillspectorState from . import static_runner -from .common import SourceLocationIndex, get_context, get_line_number, is_code_example +from .common import SourceLocationIndex, get_context_from_lines, is_code_example from .pattern_defaults import PatternCategory logger = get_logger(__name__) @@ -188,14 +188,14 @@ ) _BENIGN_AR_SCHEMA_FIELD_PATTERN = re.compile( r""" - ^\s*(?:\[\])?\s+(?:field|key|property|array|list|entry)\b + ^(?:\s*\[\])?\s+(?:field|key|property|array|list|entry)\b | - ^\s*(?:\[\])?\s+(?:in|of)\s+(?:the\s+)?(?:json(?:\s+output)?|output|response)\s+schema\b + ^(?:\s*\[\])?\s+(?:in|of)\s+(?:the\s+)?(?:json(?:\s+output)?|output|response)\s+schema\b | - ^\s*(?:\[\])?\s+(?:in|of)\s+(?:the\s+)?(?:warnings?|disclaimers?|caveats?)\b(?:\[\])?\s+ + ^(?:\s*\[\])?\s+(?:in|of)\s+(?:the\s+)?(?:warnings?|disclaimers?|caveats?)\b(?:\[\])?\s+ (?:field|key|property|array|list|entry)\b | - ^\s*(?:\[\])?\s+in\s+(?:the\s+)?errors\[\]\s+array\b + ^(?:\s*\[\])?\s+in\s+(?:the\s+)?errors\[\]\s+array\b """, re.IGNORECASE | re.VERBOSE, ) @@ -566,6 +566,7 @@ def analyze( """Analyze content for anti-refusal statements (AR1-AR3).""" findings: list[AnalyzerFinding] = [] locations = SourceLocationIndex(content, file_path) + lines = content.splitlines() tag = [PatternCategory.ANTI_REFUSAL.value] python_comments = ( _PythonComments(content, check_runtime or _no_runtime_check) @@ -584,11 +585,10 @@ def analyze( and _is_descriptive_python_comment(python_comments, match) ): continue - lines = content.splitlines() - line_num = get_line_number(content, match.start()) + line_num, column = locations.line_and_column(match.start()) match_line = lines[line_num - 1] if lines else content previous_line = lines[line_num - 2] if line_num > 1 else None - context = get_context(content, match.start(), context_lines=3) + context = get_context_from_lines(lines, line_num, window=3, column=column) security_review_context = bool( _MODE_ENABLED_RE.fullmatch(match.group(0)) and _SECURITY_REVIEW_CONTEXT_RE.search(context) 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 60423a87e..673eca50a 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_excessive_agency.py +++ b/src/skillspector/nodes/analyzers/static_patterns_excessive_agency.py @@ -424,7 +424,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()) @@ -445,7 +445,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): if (pattern, confidence) in EA2_PROSE_PATTERNS and ( @@ -495,7 +495,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..af8e90547 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py +++ b/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py @@ -80,8 +80,9 @@ ] # MP2: Context Window Stuffing — filling context to displace content +_BOUNDED_REPETITION_PATTERN = r"(.{2,20}?)\1{20,}" MP2_CONTENT_PATTERNS = [ - (r"(.{2,20}?)\1{20,}", 0.8), + (_BOUNDED_REPETITION_PATTERN, 0.8), ] MP2_PROSE_PATTERNS = [ ( @@ -541,10 +542,16 @@ def ctx(start: int) -> str: ) ) for pattern, confidence in MP2_PATTERNS: + # For each start there are only 19 candidate lengths. A failed candidate + # checks at most 20 copies; a success consumes its entire repeated run. + # There is no trailing condition to force it to backtrack over that run. + # Keep this linear, window-bounded rule on Python's faster native engine. matches = ( - static_runner.iter_paragraph_matches + re.finditer + if pattern == _BOUNDED_REPETITION_PATTERN + else 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_output_handling.py b/src/skillspector/nodes/analyzers/static_patterns_output_handling.py index 7fbdc8ca7..dd7eedd83 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_output_handling.py +++ b/src/skillspector/nodes/analyzers/static_patterns_output_handling.py @@ -634,7 +634,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in OH1_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): if pattern == _EXEC_OUTPUT_PATTERN and _is_javascript_regexp_literal_exec( @@ -690,7 +690,7 @@ def ctx(start: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in OH3_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_privilege_escalation.py b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py index 0ee0e2e23..ed5b3df5f 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py +++ b/src/skillspector/nodes/analyzers/static_patterns_privilege_escalation.py @@ -107,7 +107,7 @@ # unmatched `-- ` tokens that made the match exponential. `[\w=-]*` already # absorbs the second dash of a long option, so `-` alone covers both. ( - r"chmod\s+(?:-[\w=-]*[ \t]+)*[\"']?[+=]?0*[2-7][0-7]{3}[\"']?(?![0-9\w])", + r"chmod\s+(?>(?:-[\w=-]*[ \t]+)*)[\"']?[+=]?0*[2-7][0-7]{3}[\"']?(?![0-9\w])", 0.8, ), ] @@ -1046,7 +1046,7 @@ def context_at(offset: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in PE1_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()) @@ -1068,7 +1068,7 @@ def context_at(offset: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in PE2_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()) @@ -1093,7 +1093,7 @@ def context_at(offset: int) -> str: matches = ( static_runner.iter_paragraph_matches if (pattern, confidence) in PE3_PROSE_PATTERNS - else re.finditer + else static_runner.iter_pattern_matches ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): if _is_bare_credential_store_noun( @@ -1212,7 +1212,9 @@ def context_at(offset: int) -> str: # that match multiple patterns (e.g. DockerClient(base_url=".../docker.sock")). pe4_best: dict[int, AnalyzerFinding] = {} for pattern, confidence in PE4_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()) context = context_at(match.start()) finding_tags = list(tag) @@ -1236,7 +1238,9 @@ def context_at(offset: int) -> str: # often matches multiple flags (e.g. --privileged + --cap-add=SYS_ADMIN). pe5_best: dict[int, AnalyzerFinding] = {} for pattern, confidence in PE5_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()) context = context_at(match.start()) finding_tags = list(tag) 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 023f2741c..0a94d62f7 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_rogue_agent.py +++ b/src/skillspector/nodes/analyzers/static_patterns_rogue_agent.py @@ -273,7 +273,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()) @@ -323,7 +323,9 @@ def ctx(start: int) -> str: ) for pattern, confidence in RA2_PATTERNS: is_prose = (pattern, confidence) in RA2_PROSE_PATTERNS - matches = static_runner.iter_paragraph_matches if is_prose else re.finditer + matches = ( + static_runner.iter_paragraph_matches if is_prose else static_runner.iter_pattern_matches + ) for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): if is_prose and is_directly_prohibited(content, match.start(), match.end()): continue diff --git a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py index c2d90298b..3d0abe467 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py +++ b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py @@ -406,22 +406,23 @@ def _decoded_shell_literal_payloads(content: str) -> list[tuple[int, str]]: SC3_PATTERNS = SC3_CODE_PATTERNS + SC3_PROSE_PATTERNS # Shell decode-to-execute chains run on every file type, like SC2, because # SKILL.md instructions and hook configs carry shell commands too. +# These two exact rules retain the audited bounded native path. Additional +# shell rules default to the interruptible matcher until separately reviewed. +_SC3_BOUNDED_PIPE_PATTERN = ( + rf"{_SC3_SHELL_DECODER}(?:{_SC3_HERE_STRING}\\?['\"]?" + rf"[A-Za-z0-9+/=]{{0,{_MAX_SHELL_LITERAL_CHARS}}}+)?+{_SC3_PIPE_SEGMENT}" + rf"(?:\|(?!\|)&?{_SC3_PIPE_SEGMENT}){{0,3}}?{_SC3_PIPE_EXECUTOR}" +) +_SC3_BOUNDED_SUBSTITUTION_PATTERN = ( + rf"{_SC3_SUBSTITUTION_EXECUTOR}[^()`\n]{{0,{_MAX_SHELL_LITERAL_CHARS}}}?" + rf"{_SC3_SHELL_DECODER}" +) SC3_SHELL_PATTERNS = [ - ( - # A here-string literal can outgrow a pipeline stage, so it is consumed - # whole, and possessively, before the first stage. - rf"{_SC3_SHELL_DECODER}(?:{_SC3_HERE_STRING}\\?['\"]?" - rf"[A-Za-z0-9+/=]{{0,{_MAX_SHELL_LITERAL_CHARS}}}+)?+{_SC3_PIPE_SEGMENT}" - rf"(?:\|(?!\|)&?{_SC3_PIPE_SEGMENT}){{0,3}}?{_SC3_PIPE_EXECUTOR}", - 0.9, - ), - ( - rf"{_SC3_SUBSTITUTION_EXECUTOR}[^()`\n]{{0,{_MAX_SHELL_LITERAL_CHARS}}}?" - rf"{_SC3_SHELL_DECODER}", - 0.9, - ), + (_SC3_BOUNDED_PIPE_PATTERN, 0.9), + (_SC3_BOUNDED_SUBSTITUTION_PATTERN, 0.9), ] + # SC7: Untrusted Container Image — pulling images with signature/registry # verification turned off. These flags disable image trust regardless of the # registry, so they are a strong supply-chain signal with near-zero FP. @@ -2121,7 +2122,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 @@ -2163,7 +2166,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()) @@ -2188,7 +2191,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) @@ -2299,10 +2304,15 @@ def line_number(start: int) -> int: if has_shell_decoder: sc3_patterns = sc3_patterns + SC3_SHELL_PATTERNS for pattern, confidence in sc3_patterns: + # Shell-decoder rules already bound every stage/literal and use atomic + # or possessive ambiguous groups. Keep their audited native matcher; + # the interruptible engine adds enough cost to reject dense input. matches = ( - static_runner.iter_paragraph_matches + re.finditer + if pattern in (_SC3_BOUNDED_PIPE_PATTERN, _SC3_BOUNDED_SUBSTITUTION_PATTERN) + else 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()) @@ -2321,7 +2331,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 a0ef03d94..86fbfd548 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py +++ b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py @@ -41,6 +41,7 @@ normalized_security_view, security_text_views, ) +from skillspector.inspection_ledger import LedgerReason from skillspector.logging_config import get_logger from skillspector.models import AnalyzerFinding, Finding, Location, Severity from skillspector.python_ast import ParsedPythonFile, parse_python_source @@ -55,7 +56,9 @@ LOGICAL_LINE_BREAK, MARKDOWN_FENCE_CLOSE, MARKDOWN_FENCE_OPEN, + SourceLocationIndex, get_context, + get_context_from_lines, is_reference_material, ) from .pattern_defaults import PatternCategory @@ -221,6 +224,89 @@ r"(?:subprocess\.\w+|Popen)\s*\([^)]*\bshell\s*=\s*\1\b" ) _VARIABLE_SHELL_FLAG_RE = re.compile(_VARIABLE_SHELL_FLAG_PATTERN, re.IGNORECASE | re.MULTILINE) +_VARIABLE_SHELL_HEADER_RE = re.compile( + r"(?m)^[^\S\n]*[A-Za-z_]\w*\s*=\s*True(?=\s*$)", re.IGNORECASE +) +_VARIABLE_SHELL_PREFIX_RE = re.compile( + _VARIABLE_SHELL_FLAG_PATTERN.split(r"[^)]*", 1)[0], re.IGNORECASE | re.MULTILINE +) +# Greedy line selection also accounts for later eligible calls in the same +# assignment window. The final native match retains the original lazy order. +_VARIABLE_SHELL_LAST_PREFIX_RE = re.compile( + _VARIABLE_SHELL_PREFIX_RE.pattern.replace("{0,4}?", "{0,4}"), + re.IGNORECASE | re.MULTILINE, +) +_VARIABLE_SHELL_ARGUMENT_CHARS = 4096 + + +def _iter_variable_shell_assignments(content: str) -> Iterator[re.Match[str]]: + """Find assignment starts without retrying leading whitespace at every line.""" + for header in _VARIABLE_SHELL_HEADER_RE.finditer(content): + whitespace_start = header.start() + while whitespace_start > 0 and content[whitespace_start - 1].isspace(): + whitespace_start -= 1 + newline = content.find("\n", whitespace_start, header.start()) + start = 0 if whitespace_start == 0 else newline + 1 if newline >= 0 else header.start() + assignment = _VARIABLE_SHELL_ASSIGNMENT_RE.match(content, start) + if assignment is not None: + yield assignment + + +def _iter_variable_shell_flag_matches( + content: str, + *, + allowance: static_runner._PatternAllowance | None = None, + on_limit: Callable[[static_runner._StaticResourceLimitError], None] | None = None, +) -> Iterator[re.Match[str]]: + """Keep Python's case-folded backreferences within bounded native searches.""" + headers = _iter_variable_shell_assignments(content) + allowance = allowance or static_runner._PatternAllowance() + budget = static_runner._ACTIVE_FINDING_BUDGET.get() + consumed = 0 + while True: + if budget is not None: + budget.check_runtime() + allowance.check() + started_at = time.thread_time() + match = None + truncated = False + try: + header = next(headers, None) + if header is None: + return + if header.start() < consumed: + continue + start = max(consumed, header.start()) + prefix = _VARIABLE_SHELL_LAST_PREFIX_RE.match(content, start) + if prefix is None: + continue + end = min(len(content), prefix.end() + _VARIABLE_SHELL_ARGUMENT_CHARS) + match = _VARIABLE_SHELL_FLAG_RE.match(content, start, end) + # An artificial end must not make a partial identifier look like + # the backreference's trailing word boundary. + if match is not None and match.end() == end and end < len(content): + if content[end].isalnum() or content[end] == "_": + match = None + truncated = end < len(content) and content.find(")", prefix.end(), end) < 0 + finally: + allowance.seconds += max(0.0, time.thread_time() - started_at) + if match is not None: + consumed = match.end() + yield match + if truncated: + error = static_runner._StaticResourceLimitError( + LedgerReason.STATIC_PARSE_LIMIT, + { + "observed_characters": _VARIABLE_SHELL_ARGUMENT_CHARS + 1, + "limit_characters": _VARIABLE_SHELL_ARGUMENT_CHARS, + }, + ) + if on_limit is None: + raise error + on_limit(error) + allowance.check() + + _VARIABLE_SHELL_ASSIGNMENT_RE = re.compile( _VARIABLE_SHELL_ASSIGNMENT_PATTERN, re.IGNORECASE | re.MULTILINE ) @@ -282,7 +368,7 @@ # privilege-escalation rule: `--?` made the match exponential on a line of # unmatched `-- ` tokens, and `[\w=-]*` already absorbs a long option's # second dash. - (r"chmod\s+(?:-[\w=-]*[ \t]+)*[\"']?[+=]?0*[0-7]{2,3}[2367](?![0-9\w])[\"']?", 0.8), + (r"chmod\s+(?>(?:-[\w=-]*[ \t]+)*)[\"']?[+=]?0*[0-7]{2,3}[2367](?![0-9\w])[\"']?", 0.8), (r"(?:chmod|chown)\s+[^|]*a\+rwx", 0.8), # Git force operations (r"git\s+push\s+[^|]*--force", 0.7), @@ -4153,38 +4239,84 @@ def _is_true_prefixed_name(name: str) -> bool: return name.casefold().startswith("true") -def _variable_shell_matches(content: str) -> list[re.Match[str]]: - """Return bounded variable-shell windows, one per call for true-prefixed names. +def _variable_shell_matches( + content: str, + *, + on_limit: Callable[[static_runner._StaticResourceLimitError], None] | None = None, +) -> list[re.Match[str]]: + """Return variable-shell windows under one allowance per content window. - The non-overlapping scan pairs an assignment with one call and skips an - assignment inside an earlier window. The ``\\b``-bounded direct pattern - does not report ``shell=true_value`` and folds identical same-line - ``shell=true`` calls into one candidate, so every assignment of a - true-prefixed name is also paired with each call its bounded window - reaches. The nearest assignment owns each call. + Whole-artifact reconciliation receives the same allowance for each raw + window it covers. A caller can defer local exhaustion until other rules + have run, retaining the matches already found. """ windows: dict[int, re.Match[str]] = {} + window_count = max( + 1, + (len(content) + static_runner.SECURITY_VIEW_WINDOW_CHARS - 1) + // static_runner.SECURITY_VIEW_WINDOW_CHARS, + ) + allowance = static_runner._PatternAllowance( + limit=static_runner._STATIC_PATTERN_SECONDS * window_count + ) + budget = static_runner._ACTIVE_FINDING_BUDGET.get() def keep(window: re.Match[str]) -> None: previous = windows.get(window.end()) if previous is None or window.start() > previous.start(): windows[window.end()] = window - for match in _VARIABLE_SHELL_FLAG_RE.finditer(content): - keep(match) - for assignment in _VARIABLE_SHELL_ASSIGNMENT_RE.finditer(content): - if not _is_true_prefixed_name(assignment.group(1)): - continue - search_end = assignment.end() - 1 - for _ in range(5): - search_end = content.find("\n", search_end + 1) - if search_end == -1: - search_end = len(content) - break - for argument in _SHELL_NAME_ARGUMENT_RE.finditer(content, assignment.end(), search_end): - window = _VARIABLE_SHELL_FLAG_RE.fullmatch(content, assignment.start(), argument.end()) - if window is not None: - keep(window) + started_at = time.thread_time() + + def check_runtime() -> None: + if budget is not None: + budget.check_runtime() + allowance.seconds = max(0.0, time.thread_time() - started_at) + allowance.check() + + try: + for match in _iter_variable_shell_flag_matches( + content, allowance=allowance, on_limit=on_limit + ): + keep(match) + for assignment in _iter_variable_shell_assignments(content): + check_runtime() + if not _is_true_prefixed_name(assignment.group(1)): + continue + search_end = assignment.end() - 1 + for _ in range(5): + search_end = content.find("\n", search_end + 1) + if search_end == -1: + search_end = len(content) + break + prefix = _VARIABLE_SHELL_LAST_PREFIX_RE.match(content, assignment.start()) + if prefix is None: + continue + for argument in _SHELL_NAME_ARGUMENT_RE.finditer(content, assignment.end(), search_end): + check_runtime() + if argument.end() - prefix.end() > _VARIABLE_SHELL_ARGUMENT_CHARS: + error = static_runner._StaticResourceLimitError( + LedgerReason.STATIC_PARSE_LIMIT, + { + "observed_characters": argument.end() - prefix.end(), + "limit_characters": _VARIABLE_SHELL_ARGUMENT_CHARS, + }, + ) + if on_limit is None: + raise error + on_limit(error) + continue + window = _VARIABLE_SHELL_FLAG_RE.fullmatch( + content, assignment.start(), argument.end() + ) + check_runtime() + if window is not None: + keep(window) + check_runtime() + except static_runner._StaticResourceLimitError as error: + if on_limit is None: + raise + on_limit(error) return sorted(windows.values(), key=lambda window: (window.start(), window.end())) @@ -4417,13 +4549,16 @@ def _tm1_candidates( *, direct_shell_only: bool = False, shell_content: str | None = None, + variable_matches: list[re.Match[str]] | None = None, ) -> Iterator[tuple[int, int, str, float, bool, int | None, int | None, str | None]]: def direct_matches( pattern_index: int, pattern: str, confidence: float, ) -> Iterator[tuple[int, int, re.Match[str], float]]: - 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 + ): yield match.start(), pattern_index, match, confidence def direct_candidates() -> Iterator[ @@ -4483,13 +4618,15 @@ def ordinary_matches( confidence: float, ) -> Iterator[tuple[int, int, re.Match[str], float]]: if pattern == _VARIABLE_SHELL_FLAG_PATTERN: - for match in _variable_shell_matches(content): + for match in ( + _variable_shell_matches(content) if variable_matches is None else variable_matches + ): yield match.start(), pattern_index, match, confidence return 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(), pattern_index, match, confidence @@ -5108,14 +5245,14 @@ def analyze( ) -> list[AnalyzerFinding]: """Analyze content for tool misuse patterns (TM1–TM3).""" findings: list[AnalyzerFinding] = [] + locations = SourceLocationIndex(content, file_path) + lines = content.splitlines() def loc(ln: int) -> Location: return Location(file=file_path, start_line=ln) - line_starts = (0, *(match.end() for match in LOGICAL_LINE_BREAK.finditer(content))) - def line_number(start: int) -> int: - return bisect_right(line_starts, start) + return locations.line_and_column(start)[0] context_by_line: dict[int, str] = {} @@ -5123,7 +5260,8 @@ def ctx(start: int) -> str: line = line_number(start) context = context_by_line.get(line) if context is None: - context = get_context(content, start) + _, column = locations.line_and_column(start) + context = get_context_from_lines(lines, line, column=column) context_by_line[line] = context return context @@ -5131,12 +5269,13 @@ def ctx(start: int) -> str: tm1_findings_by_key: dict[tuple[int, str], AnalyzerFinding] = {} direct_safety_by_line: dict[int, tuple[bool, bool]] = {} + variable_limits: list[static_runner._StaticResourceLimitError] = [] variable_matches = ( {} if _direct_shell_only else { (match.start(), match.end()): (match.group(1), match) - for match in _variable_shell_matches(content) + for match in _variable_shell_matches(content, on_limit=variable_limits.append) } ) invisible_variable_matches: set[tuple[int, int]] = set() @@ -5171,6 +5310,7 @@ def ctx(start: int) -> str: content, direct_shell_only=_direct_shell_only, shell_content=shell_content, + variable_matches=[match for _, match in variable_matches.values()], ): variable_match = variable_matches.get((match_start, match_end)) if variable_match is not None and (match_start, match_end) in invisible_variable_matches: @@ -5314,7 +5454,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 = line_number(match.start()) @@ -5344,7 +5484,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 ) skip_assigned = (pattern, confidence) == _TM3_UMASK_CALL for match in matches(pattern, content, re.IGNORECASE | re.MULTILINE): @@ -5371,7 +5511,9 @@ def ctx(start: int) -> str: reference_material = is_reference_material(file_path, file_type) tm4_tags = [*tag, "contextual-triage", "likely-benign-context"] if reference_material else tag 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 = line_number(match.start()) findings.append( AnalyzerFinding( @@ -5386,6 +5528,8 @@ def ctx(start: int) -> str: complete_match=match.group(0), ) ) + if variable_limits: + raise variable_limits[0] return findings diff --git a/src/skillspector/nodes/analyzers/static_runner.py b/src/skillspector/nodes/analyzers/static_runner.py index fe3b31e1e..68234d256 100644 --- a/src/skillspector/nodes/analyzers/static_runner.py +++ b/src/skillspector/nodes/analyzers/static_runner.py @@ -17,11 +17,14 @@ from __future__ import annotations +import functools import inspect import json import math import os import re +import sys +import threading import time import unicodedata from array import array @@ -33,6 +36,8 @@ from itertools import chain from typing import cast +import regex # type: ignore[import-untyped] + from skillspector.artifacts import ( ContentKind, SecurityTextView, @@ -322,17 +327,497 @@ 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 +) + + +_CATEGORY_LOCK = threading.Lock() + + +def _python_categories(ascii_only: bool, ascii_content: bool) -> dict[str, str]: + # lru_cache alone permits duplicate builds when analyzer threads miss together. + with _CATEGORY_LOCK: + return _cached_python_categories(ascii_only, ascii_content) + + +@functools.lru_cache(maxsize=4) +def _cached_python_categories(ascii_only: bool, ascii_content: bool) -> dict[str, str]: + """Use native properties with small corrections for Python's Unicode version.""" + ascii_alphabet = ascii_only or ascii_content + alphabet = "".join(map(chr, range(128 if ascii_alphabet else sys.maxunicode + 1))) + native_classes = {"w": r"\p{L}\p{N}_", "s": r"\p{White_Space}", "d": r"\p{Nd}"} + result: dict[str, str] = {} + budget = _ACTIVE_FINDING_BUDGET.get() + + def ranges_text(ranges: list[tuple[int, int]]) -> str: + return "".join( + rf"\U{start:08x}" if start == end else rf"\U{start:08x}-\U{end:08x}" + for start, end in ranges + ) + + for category, native_class in native_classes.items(): + # These single-character classes scan the finite Unicode alphabet once, + # never attacker input. Preserve Python's membership, including its + # Unicode version and the extra ASCII whitespace separators. + native_members = bytearray(len(alphabet)) + if not ascii_alphabet: + for match in regex.finditer(f"[{native_class}]+", alphabet): + native_members[match.start() : match.end()] = b"\1" * len(match.group()) + additions: list[tuple[int, int]] = [] + removals: list[tuple[int, int]] = [] + for value, character in enumerate(alphabet): + if budget is not None and value % 4096 == 0: + budget.check_runtime() + expected = ( + character.isalnum() or character == "_" + if category == "w" + else character.isspace() + if category == "s" + else character.isdecimal() + ) + if ascii_only and category == "s": + expected = character in " \t\n\r\f\v" + if expected == bool(native_members[value]): + continue + ranges = additions if expected else removals + if ranges and ranges[-1][1] == value - 1: + ranges[-1] = (ranges[-1][0], value) + else: + ranges.append((value, value)) + added, removed = ranges_text(additions), ranges_text(removals) + if ascii_alphabet: + result[category] = f"(?-i:[{added}])" + result[category.upper()] = f"(?-i:[^{added}])" + continue + for key, negation, include, exclude in ( + (category, "", added, removed), + (category.upper(), "^", removed, added), + ): + atom = f"[{negation}{native_class}]" + if include: + atom = f"(?:{atom}|[{include}])" + if exclude: + atom = f"(?![{exclude}]){atom}" + result[key] = f"(?-i:{atom})" + return result + + +def _requires_leading_word(source: str) -> bool: + """Prove a leading word atom for the small trusted static-pattern grammar.""" + if not source: + return False + if source[0].isascii() and (source[0].isalnum() or source[0] == "_"): + return len(source) == 1 or source[1] not in "?*{" + if not source.startswith("(?:"): + return False + branches = [] + branch_start = 3 + depth = 1 + cursor = 3 + in_class = False + while cursor < len(source): + char = source[cursor] + if char == "\\": + cursor += 2 + continue + if char == "[": + in_class = True + elif char == "]": + in_class = False + elif not in_class: + if char == "(": + depth += 1 + elif char == ")": + depth -= 1 + if depth == 0: + branches.append(source[branch_start:cursor]) + return (cursor + 1 == len(source) or source[cursor + 1] not in "?*{") and all( + _requires_leading_word(branch) for branch in branches + ) + elif char == "|" and depth == 1: + branches.append(source[branch_start:cursor]) + branch_start = cursor + 1 + cursor += 1 + return False + + +@functools.lru_cache(maxsize=1024) +def _timed_pattern(source: str, flags: int, ascii_content: bool = False) -> 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 + + # ASCII input needs only ASCII category members. Keep the original flags: + # Unicode \s includes ASCII control separators, and Unicode literals can + # still fold to ASCII letters under IGNORECASE. + categories = _python_categories(bool(flags & re.ASCII), ascii_content) + word = 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): + if source.startswith("(?P=", cursor) and flags & re.IGNORECASE and not ascii_content: + raise _StaticResourceLimitError(LedgerReason.RULES_UNAVAILABLE, {}) + 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 "123456789" and flags & re.IGNORECASE and not ascii_content: + raise _StaticResourceLimitError(LedgerReason.RULES_UNAVAILABLE, {}) + 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(categories[code]) + elif code == "b": + following = source[cursor + 2 :] + required_word = _requires_leading_word(following) + previous_word = bool( + parts + and len(parts[-1]) == 1 + and re.fullmatch(r"\w", parts[-1], flags & re.ASCII) + ) + if required_word: + parts.append(rf"(? tuple[re.Pattern[str], re.Pattern[str], re.Pattern[str]]: + if source == _LINEAR_HIDDEN_DIRECTORY_PATTERN: + wildcard = rf"(?:[^|&;{LINE_BREAK_CHARS}]|\\\r?\n)*" + # Backslash-LF and backslash-CRLF continue a shell logical line. Other + # command separators and line breaks end the rule's wildcard span. + other_breaks = re.escape(LINE_BREAK_CHARS.replace("\r", "").replace("\n", "")) + separator = rf"[|&;{other_breaks}]|(? Iterator[re.Match[str]]: + """Preserve greedy matches while visiting each delimiter-free segment once. + + The wildcard chooses the rightmost possible suffix. If an earlier command + precedes it, the original pattern can match at that command in one anchored + pass. Without such a pair, skip the segment without retrying every command. + Audited suffixes may consume delimiters, as in a download followed by && sh. + Each successful match consumes its segment's last suffix; the total prefix, + suffix and anchored scanning is linear in the window length. + """ + start, end = max(0, start), min(len(content), max(0, end)) + prefix, suffix, separator = _linear_command_parts(original.pattern, original.flags) + suffixes = suffix.finditer(content, start, end) + candidate = next(suffixes, None) + budget = _ACTIVE_FINDING_BUDGET.get() + while start < end: + if budget is not None: + budget.check_runtime() + delimiter = separator.search(content, start, end) + stop = end if delimiter is None else delimiter.start() + last_suffix = None + while candidate is not None and candidate.start() <= stop: + if candidate.start() >= start: + last_suffix = candidate + candidate = next(suffixes, None) + if last_suffix is not None: + command = prefix.search(content, start, last_suffix.start()) + if command is not None: + match = original.match(content, command.start(), end) + if match is None: + raise _StaticResourceLimitError(LedgerReason.RULES_UNAVAILABLE, {}) + start = max(stop + 1, match.end()) + yield match + continue + start = stop + 1 + + +@dataclass +class _PatternAllowance: + seconds: float = 0.0 + limit: float = field(default_factory=lambda: _STATIC_PATTERN_SECONDS) + + def check(self) -> None: + if self.seconds >= self.limit: + raise _StaticResourceLimitError( + LedgerReason.RUNTIME_LIMIT, + {"observed_seconds": self.seconds, "limit_seconds": self.limit}, + ) + + +def _iter_pattern_matches( + original: re.Pattern[str], + compiled: regex.Pattern[str] | None, + content: str, + start: int, + end: int, + allowance: _PatternAllowance, + budget: _FindingBudget | None, +) -> Iterator[re.Match[str]]: + linear_matches = ( + _linear_command_matches(original, content, start, end) if compiled is None else None + ) + skip_empty = False + try: + while True: + if budget is not None: + budget.check_runtime() + allowance.limit = min( + allowance.limit, + allowance.seconds + max(0.0, budget.deadline - budget.clock()), + ) + timeout = allowance.limit - allowance.seconds + if timeout <= 0: + raise TimeoutError + # regex's iterator timer includes CPU used by the caller between + # yields. Restart it while retaining one cumulative matching budget; + # the artifact deadline separately bounds caller work. + started_at = time.thread_time() + expired = False + try: + if linear_matches is not None: + match = next(linear_matches, None) + else: + assert compiled is not None + matches = compiled.finditer( + content, + start, + len(content) if end is None else end, + timeout=timeout, + concurrent=False, + ) + if skip_empty: + # Replay the previous empty match so the native iterator + # can still return a nonempty match at the same position. + next(matches, None) + match = next(matches, None) + except TimeoutError: + expired = True + finally: + allowance.seconds += max(0.0, time.thread_time() - started_at) + if budget is not None: + budget.check_runtime() + if linear_matches is not None and allowance.seconds >= allowance.limit: + raise TimeoutError + if expired: + # The engine counts all process CPU. Retry within this thread's + # allowance; sustained contention can still exhaust it through + # repeated searches that make no retained progress. + continue + if match is None: + return + start = match.end() + skip_empty = match.start() == start + yield cast(re.Match[str], match) + except TimeoutError as exc: + raise _StaticResourceLimitError( + LedgerReason.RUNTIME_LIMIT, + { + "observed_seconds": allowance.seconds, + "limit_seconds": allowance.limit, + }, + ) from exc + + +def iter_pattern_matches( + pattern: str | re.Pattern[str], + content: str, + flags: int = 0, + *, + start: int = 0, + end: int | None = None, +) -> 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 = ( + None + if original.pattern in _LINEAR_COMMAND_PATTERNS and not original.flags & re.VERBOSE + else _timed_pattern(original.pattern, original.flags, content.isascii()) + ) + yield from _iter_pattern_matches( + original, + compiled, + content, + start, + len(content) if end is None else end, + _PatternAllowance(), + budget, + ) + + 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) + """Match prose ranges with one matching allowance per rule and content window.""" 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) + 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, content.isascii()) + allowance = _PatternAllowance() + cursor = 0 + while cursor < len(ranges): + if budget is not None: + budget.check_runtime() + allowance.check() + start, end = ranges[cursor] + if end - start > 8: + yield from _iter_pattern_matches( + original, compiled, content, start, end, allowance, budget + ) + cursor += 1 + continue + # At most eight characters per native search bounds backtracking in the + # trusted rule catalog. Batch these tiny ranges to avoid a timer/engine + # setup for each paragraph; collect before yielding to exclude caller CPU. + started_at = time.thread_time() + matches = [] + for _ in range(128): + start, end = ranges[cursor] + if end - start > 8: + break + matches.extend(original.finditer(content, start, end)) + cursor += 1 + if cursor == len(ranges): + break + allowance.seconds += max(0.0, time.thread_time() - started_at) + allowance.check() + for match in matches: + if budget is not None: + budget.check_runtime() + yield match def security_view_match_is_literal(content: str, start: int, end: int) -> bool: @@ -998,6 +1483,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)) @@ -1056,6 +1542,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 @@ -3428,11 +3916,12 @@ def run_static_patterns_with_ledger( or observed_seconds >= runtime_limit ) if expired: + if resource_limit is not LedgerReason.RUNTIME_LIMIT: + resource_metrics = { + "observed_seconds": observed_seconds, + "limit_seconds": runtime_limit, + } resource_limit = LedgerReason.RUNTIME_LIMIT - resource_metrics = { - "observed_seconds": observed_seconds, - "limit_seconds": runtime_limit, - } path_findings = _cleanup_expired_path_findings( pattern_modules, path_findings, @@ -3486,7 +3975,9 @@ def run_static_patterns_with_ledger( resource_limit = None resource_metrics = {} except _StaticResourceLimitError as exc: - path_findings = [] + path_findings = _cleanup_expired_path_findings( + pattern_modules, path_findings + ) resource_limit = exc.reason resource_metrics = exc.metrics except Exception as exc: @@ -3533,10 +4024,18 @@ def run_static_patterns_with_ledger( reason=partial_reason if partial else None, emitted_finding_ids=[finding.finding_id for finding in path_findings], observed_characters=( - len(content) if partial_reason is LedgerReason.SIZE_LIMIT else None + int(resource_metrics["observed_characters"]) + if "observed_characters" in resource_metrics + else len(content) + if partial_reason is LedgerReason.SIZE_LIMIT + else None ), limit_characters=( - MAX_FILE_CHARS if partial_reason is LedgerReason.SIZE_LIMIT else None + int(resource_metrics["limit_characters"]) + if "limit_characters" in resource_metrics + else MAX_FILE_CHARS + if partial_reason is LedgerReason.SIZE_LIMIT + else None ), observed_findings=( int(resource_metrics.get("observed_findings", len(path_findings))) diff --git a/tests/nodes/analyzers/test_static_patterns.py b/tests/nodes/analyzers/test_static_patterns.py index cbd2aa2f0..59a21b697 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 @@ -2249,3 +2250,956 @@ 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"ignore\s+previous", "ignore\x1cprevious", re.IGNORECASE), + (r"ignore\s+previous", "ignore\x1cprevious", re.IGNORECASE | re.ASCII), + (r"\bK\w+\b", "key", re.IGNORECASE), + (r"\bſ\w+\b", "safe", re.IGNORECASE), + (r"[\W\d]+", "a1 !\x1c", 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 + + +@pytest.mark.parametrize("pattern", [r".*?", r"a*|b", r"\b", r"a?", r"(a)?b|", r"(a)?\1|b|"]) +@pytest.mark.parametrize(("start", "end"), [(0, 4), (1, 3), (2, 2)]) +def test_timed_patterns_preserve_empty_match_adjacency(pattern, start, end): + text = "abab" + expected = [ + (match.span(), match.group(), match.groups()) + for match in re.compile(pattern).finditer(text, start, end) + ] + actual = [ + (match.span(), match.group(), match.groups()) + for match in static_runner.iter_pattern_matches(pattern, text, start=start, end=end) + ] + assert actual == expected + + +def test_timed_pattern_does_not_charge_consumer_cpu(monkeypatch): + monkeypatch.setattr(static_runner, "_STATIC_PATTERN_SECONDS", 0.05) + matches = static_runner.iter_pattern_matches(r".", "abc") + assert next(matches).group() == "a" + deadline = static_runner.time.process_time() + 0.1 + while static_runner.time.process_time() < deadline: + pass + assert [match.group() for match in matches] == ["b", "c"] + + +def test_timed_pattern_rechecks_artifact_deadline_after_consumer(): + now = 0.0 + budget = static_runner._FindingBudget(10, 0.0, 1.0, lambda: now) + token = static_runner._ACTIVE_FINDING_BUDGET.set(budget) + try: + matches = static_runner.iter_pattern_matches(r".", "abc") + assert next(matches).group() == "a" + now = 2.0 + with pytest.raises(static_runner._StaticResourceLimitError) as caught: + next(matches) + assert caught.value.reason.value == "runtime_limit" + finally: + static_runner._ACTIVE_FINDING_BUDGET.reset(token) + + +def test_static_regex_deadline_retains_findings_and_incomplete_ledger(monkeypatch): + original = static_runner._timed_pattern + + def compile_with_small_comment_deadline(source, flags, ascii_content=False): + if source.startswith("", "