Use byte offsets for YARA line lookup

src/skillspector/nodes/analyzers/static_yara.py

@@ -40,7 +40,7 @@
 from skillspector.models import AnalyzerFinding, Location, Severity
 from skillspector.state import AnalyzerNodeResponse, SkillspectorState
 
-from .common import get_context, get_line_number
+from .common import get_context_from_lines
 from .pattern_defaults import PatternCategory
 from .static_runner import MAX_FILE_CHARS, analyzer_finding_to_finding
 
@@ -194,20 +194,25 @@ def _load_rules(extra_dir: Path | None = None) -> yara.Rules | None:
 
 def _extract_match_strings(match: yara.Match) -> tuple[int, str | None]:
     """Extract the first match offset and a joined matched-text snippet from a YARA match."""
-    first_offset = 0
+    first_offset: int | None = None
     parts: list[str] = []
     for sd in match.strings or []:
         for inst in sd.instances or []:
-            if first_offset == 0:
+            if first_offset is None or inst.offset < first_offset:
                 first_offset = inst.offset
             matched_bytes = inst.matched_data
             if isinstance(matched_bytes, bytes):
                 parts.append(matched_bytes.decode("utf-8", errors="replace"))
     matched_text = "; ".join(parts)[:200] if parts else None
-    return first_offset, matched_text
+    return first_offset if first_offset is not None else 0, matched_text
 
 
-def _has_local_destructive_autonomy_evidence(match: yara.Match, content: str) -> bool:
+def _line_number_from_byte_offset(data: bytes, offset: int) -> int:
+    """Return the 1-based line number for a YARA byte offset in *data*."""
+    return data[:offset].count(b"\n") + 1
+
+
+def _has_local_destructive_autonomy_evidence(match: yara.Match, data: bytes) -> bool:
     """Require destructive and autonomy evidence to occur in one local context.
 
     YARA string conditions are file-wide. Without this post-match check, a
@@ -221,7 +226,7 @@ def _has_local_destructive_autonomy_evidence(match: yara.Match, content: str) ->
     for string_match in match.strings or []:
         identifier = str(string_match.identifier)
         for instance in string_match.instances or []:
-            line = get_line_number(content, instance.offset)
+            line = _line_number_from_byte_offset(data, instance.offset)
             if identifier == "$destructive_rm_root":
                 return True
             if identifier.startswith("$destructive_"):
@@ -274,7 +279,7 @@ def _match_file(rules: yara.Rules, content: str, file_path: str) -> list[Analyze
     for match in matches:
         if (
             match.rule == _DESTRUCTIVE_AUTONOMY_RULE
-            and not _has_local_destructive_autonomy_evidence(match, content)
+            and not _has_local_destructive_autonomy_evidence(match, data)
         ):
             logger.debug(
                 "%s: ignored cross-context destructive/autonomy match in %s",
@@ -284,18 +289,17 @@ def _match_file(rules: yara.Rules, content: str, file_path: str) -> list[Analyze
             continue
         rule_id, severity, confidence, description = _parse_meta(match)
         first_offset, matched_text = _extract_match_strings(match)
+        start_line = _line_number_from_byte_offset(data, first_offset)
 
         findings.append(
             AnalyzerFinding(
                 rule_id=rule_id,
                 message=_build_message(match.rule, match.namespace, description),
                 severity=severity,
-                location=Location(
-                    file=file_path, start_line=get_line_number(content, first_offset)
-                ),
+                location=Location(file=file_path, start_line=start_line),
                 confidence=confidence,
                 tags=[PatternCategory.YARA_MATCH.value],
-                context=get_context(content, first_offset),
+                context=get_context_from_lines(content.splitlines(), start_line),
                 matched_text=matched_text,
             )
         )

tests/nodes/analyzers/test_static_yara.py

@@ -140,6 +140,39 @@ def test_finding_fields_populated(self, tmp_path):
         assert "YARA Match" in f.tags
         assert f.remediation is not None
 
+    def test_multibyte_prefix_preserves_finding_line_and_context(self, tmp_path):
+        _write_rule(
+            tmp_path,
+            "detect_unicode_marker",
+            category="malware",
+            severity="HIGH",
+            strings={"a": "UNICODE_MARKER"},
+        )
+        trailing_lines = "\n".join(f"tail {index}" for index in range(10))
+        content = f"{'😀' * 50}\nline two\nUNICODE_MARKER\n{trailing_lines}"
+
+        finding = _run(content, "unicode.txt", str(tmp_path))[0]
+
+        assert finding.start_line == 3
+        assert "UNICODE_MARKER" in finding.context
+
+    def test_match_at_byte_zero_remains_the_first_offset(self, tmp_path):
+        _write_rule(
+            tmp_path,
+            "detect_multiple_markers",
+            category="malware",
+            severity="HIGH",
+            strings={"first": "START_MARKER", "later": "LATER_MARKER"},
+        )
+
+        finding = _run(
+            "START_MARKER\nmiddle line\nLATER_MARKER",
+            "multiple.txt",
+            str(tmp_path),
+        )[0]
+
+        assert finding.start_line == 1
+
     def test_message_contains_rule_name(self, tmp_path):
         _write_rule(
             tmp_path,
@@ -441,6 +474,19 @@ def test_destructive_action_does_not_combine_with_distant_autonomy_prose(self):
         findings = _run_builtin(content, "SKILL.md")
         assert not _has_rule(findings, "agent_skill_destructive_autonomous_actions")
 
+    def test_multibyte_prefix_does_not_collapse_distant_destructive_evidence(self):
+        intervening_lines = "\n".join(f"review step {index}" for index in range(6))
+        content = (
+            f"{'😀' * 100}\n"
+            "rm -rf ./workspace\n"
+            f"{intervening_lines}\n"
+            "continue without confirmation\n"
+        )
+
+        findings = _run_builtin(content, "SKILL.md")
+
+        assert not _has_rule(findings, "agent_skill_destructive_autonomous_actions")
+
     def test_destructive_root_delete_remains_blocking_without_autonomy_phrase(self):
         findings = _run_builtin("rm -rf /\n", "setup.sh")
         assert _has_rule(findings, "agent_skill_destructive_autonomous_actions")