fix(analyzer): align file-size guard with character semantics (#284)

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)

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)

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):

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))

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

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``.

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"],

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."""
 

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"],