diff --git a/src/skillspector/nodes/analyzers/behavioral_ast.py b/src/skillspector/nodes/analyzers/behavioral_ast.py index e571c57a..badf980a 100644 --- a/src/skillspector/nodes/analyzers/behavioral_ast.py +++ b/src/skillspector/nodes/analyzers/behavioral_ast.py @@ -30,7 +30,7 @@ resolve_call_name, resolve_dynamic_import_call, ) -from .static_runner import MAX_FILE_BYTES, analyzer_finding_to_finding +from .static_runner import MAX_FILE_CHARS, analyzer_finding_to_finding ANALYZER_ID = "behavioral_ast" logger = get_logger(__name__) @@ -243,7 +243,7 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse: if not path.endswith(".py"): continue content = file_cache.get(path) - if content is None or len(content) > MAX_FILE_BYTES: + if content is None or len(content) > MAX_FILE_CHARS: continue raw = _analyze_python(content, path) all_findings.extend(analyzer_finding_to_finding(af) for af in raw) diff --git a/src/skillspector/nodes/analyzers/behavioral_taint_tracking.py b/src/skillspector/nodes/analyzers/behavioral_taint_tracking.py index f6141337..344eae09 100644 --- a/src/skillspector/nodes/analyzers/behavioral_taint_tracking.py +++ b/src/skillspector/nodes/analyzers/behavioral_taint_tracking.py @@ -39,7 +39,7 @@ resolve_dotted_name, resolve_dynamic_import_call, ) -from .static_runner import MAX_FILE_BYTES, analyzer_finding_to_finding +from .static_runner import MAX_FILE_CHARS, analyzer_finding_to_finding ANALYZER_ID = "behavioral_taint_tracking" logger = get_logger(__name__) @@ -430,7 +430,7 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse: if not path.endswith(".py"): continue content = file_cache.get(path) - if content is None or len(content) > MAX_FILE_BYTES: + if content is None or len(content) > MAX_FILE_CHARS: continue raw = _analyze_python(content, path) all_findings.extend(analyzer_finding_to_finding(af) for af in raw) diff --git a/src/skillspector/nodes/analyzers/static_runner.py b/src/skillspector/nodes/analyzers/static_runner.py index f97ef8a4..539dc548 100644 --- a/src/skillspector/nodes/analyzers/static_runner.py +++ b/src/skillspector/nodes/analyzers/static_runner.py @@ -48,7 +48,7 @@ ".rs": "rust", } -MAX_FILE_BYTES = 1_000_000 +MAX_FILE_CHARS = 1_000_000 _EVAL_DATASET_FILES = { "evals/evals.json", "evals/evals.jsonl", @@ -272,12 +272,12 @@ def run_static_patterns( if content is None: logger.debug("Skipping %s: no content in file_cache", path) continue - if len(content) > MAX_FILE_BYTES: + if len(content) > MAX_FILE_CHARS: logger.debug( - "Skipping %s: size %d exceeds MAX_FILE_BYTES (%d)", + "Skipping %s: size %d characters exceeds MAX_FILE_CHARS (%d)", path, len(content), - MAX_FILE_BYTES, + MAX_FILE_CHARS, ) continue if _is_binary_file(path, content): diff --git a/src/skillspector/nodes/analyzers/static_yara.py b/src/skillspector/nodes/analyzers/static_yara.py index 9d753d5d..fb675ff0 100644 --- a/src/skillspector/nodes/analyzers/static_yara.py +++ b/src/skillspector/nodes/analyzers/static_yara.py @@ -35,7 +35,7 @@ from .common import get_context, get_line_number from .pattern_defaults import PatternCategory -from .static_runner import MAX_FILE_BYTES, analyzer_finding_to_finding +from .static_runner import MAX_FILE_CHARS, analyzer_finding_to_finding ANALYZER_ID = "static_yara" logger = get_logger(__name__) @@ -276,8 +276,13 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse: content = file_cache.get(path) if content is None: continue - if len(content) > MAX_FILE_BYTES: - logger.debug("%s: skipping %s (exceeds size limit)", ANALYZER_ID, path) + if len(content) > MAX_FILE_CHARS: + logger.debug( + "%s: skipping %s (exceeds %d-character limit)", + ANALYZER_ID, + path, + MAX_FILE_CHARS, + ) continue for af in _match_file(rules, content, path): findings.append(analyzer_finding_to_finding(af)) diff --git a/src/skillspector/providers/_agent_cli.py b/src/skillspector/providers/_agent_cli.py index d7aa415d..cf53f508 100644 --- a/src/skillspector/providers/_agent_cli.py +++ b/src/skillspector/providers/_agent_cli.py @@ -65,9 +65,9 @@ # Constants # --------------------------------------------------------------------------- -# Reuse the same cap as static_runner so a skill that's too big for static -# analysis is also too big to send to the CLI. -MAX_INPUT_BYTES = 1_000_000 # 1 MB — mirrors MAX_FILE_BYTES in static_runner.py +# Static analyzers stop at a one-million-character decoded text limit. +# The CLI prompt path separately caps encoded UTF-8 bytes. +MAX_INPUT_BYTES = 1_000_000 # 1 MB encoded prompt cap MAX_OUTPUT_BYTES = 10_000_000 # 10 MB safety cap on stdout MAX_STDERR_BYTES = 64_000 # stderr is only used for error snippets CLI_TIMEOUT_SECONDS = 300 # 5-minute per-call hard limit diff --git a/tests/nodes/analyzers/test_behavioral_ast.py b/tests/nodes/analyzers/test_behavioral_ast.py index ae1a4231..96af460c 100644 --- a/tests/nodes/analyzers/test_behavioral_ast.py +++ b/tests/nodes/analyzers/test_behavioral_ast.py @@ -233,6 +233,46 @@ def test_missing_file_in_cache(self): result = behavioral_ast.node(state) assert result["findings"] == [] + def test_file_size_gate_scans_exact_character_limit(self): + from skillspector.nodes.analyzers.static_runner import MAX_FILE_CHARS + + prefix = 'exec("x")\n' + code = prefix + (" " * (MAX_FILE_CHARS - len(prefix))) + assert len(code) == MAX_FILE_CHARS + assert any(f.rule_id == "AST1" for f in _run(code)) + + def test_file_size_gate_skips_over_character_limit(self): + from skillspector.nodes.analyzers.static_runner import MAX_FILE_CHARS + + prefix = 'exec("x")\n' + code = prefix + (" " * (MAX_FILE_CHARS - len(prefix) + 1)) + assert len(code) == MAX_FILE_CHARS + 1 + assert _run(code) == [] + + def test_file_size_gate_multibyte_under_character_limit_scanned(self): + from skillspector.nodes.analyzers.static_runner import MAX_FILE_CHARS + + prefix = 'exec("x")\n# ' + code = prefix + ("🦄" * 250_000) + assert len(code) <= MAX_FILE_CHARS + assert len(code.encode("utf-8")) > MAX_FILE_CHARS + assert any(f.rule_id == "AST1" for f in _run(code)) + + def test_file_size_gate_skips_only_oversized_component(self): + from skillspector.nodes.analyzers.static_runner import MAX_FILE_CHARS + + big = 'exec("x")\n' + (" " * MAX_FILE_CHARS) + small = 'exec("ok")\n' + state = { + "components": ["big.py", "small.py"], + "file_cache": {"big.py": big, "small.py": small}, + } + + result = behavioral_ast.node(state) + files = {f.file for f in result["findings"]} + assert "big.py" not in files + assert "small.py" in files + class TestImportAliasEvasion: """Dangerous calls must be detected through ``from ... import`` and ``import ... as``. diff --git a/tests/nodes/analyzers/test_behavioral_taint_tracking.py b/tests/nodes/analyzers/test_behavioral_taint_tracking.py index 699396be..1238050e 100644 --- a/tests/nodes/analyzers/test_behavioral_taint_tracking.py +++ b/tests/nodes/analyzers/test_behavioral_taint_tracking.py @@ -259,13 +259,45 @@ def test_missing_file_in_cache(self): assert result["findings"] == [] def test_oversized_file_skipped(self): - from skillspector.nodes.analyzers.static_runner import MAX_FILE_BYTES + from skillspector.nodes.analyzers.static_runner import MAX_FILE_CHARS - big = 'import os\nexec(os.environ.get("KEY"))\n' + ("x = 1\n" * MAX_FILE_BYTES) + big = 'import os\nexec(os.environ.get("KEY"))\n' + ("x = 1\n" * MAX_FILE_CHARS) state = {"components": ["big.py"], "file_cache": {"big.py": big}} result = behavioral_taint_tracking.node(state) assert result["findings"] == [] + def test_exact_character_limit_scanned(self): + from skillspector.nodes.analyzers.static_runner import MAX_FILE_CHARS + + prefix = 'import os\nexec(os.environ.get("KEY"))\n' + code = prefix + (" " * (MAX_FILE_CHARS - len(prefix))) + assert len(code) == MAX_FILE_CHARS + assert _rule_ids(_run(code)) + + def test_multibyte_under_char_limit_scanned(self): + from skillspector.nodes.analyzers.static_runner import MAX_FILE_CHARS + + prefix = 'import os\nexec(os.environ.get("KEY"))\n# ' + code = prefix + ("🦄" * 250_000) + assert len(code) <= MAX_FILE_CHARS + assert len(code.encode("utf-8")) > MAX_FILE_CHARS + assert _rule_ids(_run(code)) + + def test_oversized_file_does_not_stop_later_components(self): + from skillspector.nodes.analyzers.static_runner import MAX_FILE_CHARS + + big = 'import os\nexec(os.environ.get("KEY"))\n' + ("x = 1\n" * MAX_FILE_CHARS) + small = 'import os\nexec(os.environ.get("KEY"))\n' + state = { + "components": ["big.py", "small.py"], + "file_cache": {"big.py": big, "small.py": small}, + } + + result = behavioral_taint_tracking.node(state) + files = {f.file for f in result["findings"]} + assert "big.py" not in files + assert "small.py" in files + def test_multiple_files_produce_findings(self): state = { "components": ["a.py", "b.py"], diff --git a/tests/nodes/analyzers/test_static_runner_filtering.py b/tests/nodes/analyzers/test_static_runner_filtering.py index a69d7870..7f5a39dc 100644 --- a/tests/nodes/analyzers/test_static_runner_filtering.py +++ b/tests/nodes/analyzers/test_static_runner_filtering.py @@ -31,6 +31,80 @@ def _findings(content: str, path: str, module: object) -> set[str]: return {finding.rule_id for finding in static_runner.run_static_patterns(state, [module])} +class _RecordingModule: + def __init__(self) -> None: + self.calls: list[str] = [] + + def analyze(self, *, content: str, file_path: str, file_type: str) -> list: + self.calls.append(content) + return [] + + +class TestCharacterLimit: + def test_char_gate_scans_at_limit_skips_above(self) -> None: + module = _RecordingModule() + limit = static_runner.MAX_FILE_CHARS + + assert ( + static_runner.run_static_patterns( + {"components": ["exact.txt"], "file_cache": {"exact.txt": "x" * limit}}, + [module], + ) + == [] + ) + assert len(module.calls) == 1 + + module.calls.clear() + assert ( + static_runner.run_static_patterns( + {"components": ["over.txt"], "file_cache": {"over.txt": "x" * (limit + 1)}}, + [module], + ) + == [] + ) + assert module.calls == [] + + def test_multibyte_under_char_limit_scanned(self) -> None: + module = _RecordingModule() + content = "🦄" * 250_001 + assert len(content) <= static_runner.MAX_FILE_CHARS + assert len(content.encode("utf-8")) > static_runner.MAX_FILE_CHARS + + static_runner.run_static_patterns( + {"components": ["unicode.txt"], "file_cache": {"unicode.txt": content}}, + [module], + ) + assert module.calls == [content] + + def test_oversized_file_does_not_stop_later_components(self) -> None: + module = _RecordingModule() + limit = static_runner.MAX_FILE_CHARS + state = { + "components": ["over.txt", "small.txt"], + "file_cache": { + "over.txt": "x" * (limit + 1), + "small.txt": "SAFE", + }, + } + + assert static_runner.run_static_patterns(state, [module]) == [] + assert module.calls == ["SAFE"] + + def test_skip_log_reports_char_metric(self, caplog) -> None: + caplog.set_level("DEBUG", logger="skillspector.nodes.analyzers.static_runner") + content = "x" * (static_runner.MAX_FILE_CHARS + 1) + + static_runner.run_static_patterns( + {"components": ["over.txt"], "file_cache": {"over.txt": content}}, + [_RecordingModule()], + ) + + message = " ".join(record.getMessage() for record in caplog.records) + assert "characters" in message + assert "MAX_FILE_CHARS" in message + assert "MAX_FILE_BYTES" not in message + + class TestSemanticStringDocumentationFiltering: """Governed lexical rules are filtered only in non-executable documentation contexts.""" diff --git a/tests/nodes/analyzers/test_static_yara.py b/tests/nodes/analyzers/test_static_yara.py index b1472a3c..c42d1012 100644 --- a/tests/nodes/analyzers/test_static_yara.py +++ b/tests/nodes/analyzers/test_static_yara.py @@ -27,7 +27,7 @@ import pytest from skillspector.nodes.analyzers import static_yara -from skillspector.nodes.analyzers.static_runner import MAX_FILE_BYTES +from skillspector.nodes.analyzers.static_runner import MAX_FILE_CHARS @pytest.fixture(autouse=True) @@ -270,10 +270,44 @@ def test_oversized_file_skipped(self, tmp_path): _write_rule( tmp_path, "rule_big", category="malware", severity="HIGH", strings={"a": "BIGMARKER"} ) - content = "BIGMARKER" + ("x" * MAX_FILE_BYTES) + content = "BIGMARKER" + ("x" * MAX_FILE_CHARS) findings = _run(content, "big.txt", str(tmp_path)) assert findings == [] + def test_exact_character_limit_scanned(self, tmp_path): + _write_rule( + tmp_path, "rule_exact", category="malware", severity="HIGH", strings={"a": "EXACT"} + ) + content = "EXACT" + ("x" * (MAX_FILE_CHARS - len("EXACT"))) + findings = _run(content, "exact.txt", str(tmp_path)) + assert _has_rule(findings, "rule_exact") + + def test_multibyte_under_char_limit_scanned(self, tmp_path): + _write_rule( + tmp_path, "rule_unicode", category="malware", severity="HIGH", strings={"a": "UNICODE"} + ) + content = "UNICODE" + ("🦄" * 250_000) + assert len(content) <= MAX_FILE_CHARS + assert len(content.encode("utf-8")) > MAX_FILE_CHARS + assert _has_rule(_run(content, "unicode.txt", str(tmp_path)), "rule_unicode") + + def test_oversized_file_does_not_stop_later_components(self, tmp_path): + _write_rule( + tmp_path, "rule_small", category="malware", severity="HIGH", strings={"a": "SMALL"} + ) + state = { + "components": ["big.txt", "small.txt"], + "file_cache": { + "big.txt": "BIGMARKER" + ("x" * MAX_FILE_CHARS), + "small.txt": "SMALL", + }, + "yara_rules_dir": str(tmp_path), + } + + findings = static_yara.node(state)["findings"] + assert _has_rule(findings, "rule_small") + assert {f.file for f in findings} == {"small.txt"} + def test_nonexistent_rules_dir_returns_empty(self): state = { "components": ["f.txt"],