fix(security): preserve exact finding identity

src/skillspector/models.py

@@ -21,7 +21,7 @@
 from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from contextvars import ContextVar
-from dataclasses import dataclass, field
+from dataclasses import InitVar, dataclass, field
 from enum import StrEnum
 from hashlib import sha256
 from typing import TYPE_CHECKING, Protocol
@@ -40,6 +40,12 @@ class Severity(StrEnum):
     CRITICAL = "CRITICAL"
 
 
+def compute_match_fingerprint(rule_id: str, matched_text: str) -> str:
+    """Return the canonical SHA-256 identity for one rule-bound match."""
+    normalized = " ".join(matched_text.strip().split())
+    return sha256(f"{rule_id}\x1f{normalized}".encode()).hexdigest()
+
+
 @dataclass
 class Location:
     """Location of a finding within a file (used by all analyzers)."""
@@ -72,8 +78,11 @@ class AnalyzerFinding:
     context: str | None = None
     matched_text: str | None = None
     evidence: dict[str, object] = field(default_factory=dict)
+    # Canonical rule+match digest; source binding is derived by ``Finding.fingerprint``.
+    match_fingerprint: str | None = None
+    complete_match: InitVar[str | None] = None
 
-    def __post_init__(self) -> None:
+    def __post_init__(self, complete_match: str | None) -> None:
         """Notify an optional runner-owned resource guard after construction.
 
         Static analyzers are trusted code, but the number of findings they
@@ -82,6 +91,8 @@ def __post_init__(self) -> None:
         private result list instead of waiting for that list to become large.
         Other analyzer families pay no cost beyond this single context lookup.
         """
+        if complete_match is not None:
+            self.match_fingerprint = compute_match_fingerprint(self.rule_id, complete_match)
         observer = _analyzer_finding_observer.get()
         if observer is not None:
             observer(self)
@@ -135,6 +146,7 @@ class Finding:
     source_identity: str | None = None
     source_digest: str | None = None
     evidence: dict[str, object] = field(default_factory=dict)
+    # Canonical unbound rule+match digest. Never replace it with a source-bound digest.
     match_fingerprint: str | None = None
     occurrences: list[dict[str, object]] = field(default_factory=list)
 
@@ -173,7 +185,11 @@ def fingerprint(self) -> str | None:
             else " ".join((self.matched_text or "").strip().split())
         )
         if not has_source_provenance:
-            return sha256(f"{self.rule_id}\x1f{normalized}".encode()).hexdigest()
+            return (
+                self.match_fingerprint
+                if self.match_fingerprint
+                else compute_match_fingerprint(self.rule_id, normalized)
+            )
         payload = {
             "rule_id": self.rule_id,
             "match": normalized,

src/skillspector/nodes/analyzers/behavioral_ast.py

@@ -40,8 +40,8 @@
 )
 
 from .common import (
+    get_complete_source_segment,
     get_context_from_lines,
-    get_source_segment,
     resolve_call_name,
     resolve_dynamic_import_call,
 )
@@ -341,10 +341,17 @@ def _analyze_python(
 
     def _emit(
         rule_id: str,
-        lineno: int,
-        end_lineno: int | None,
+        ast_node: ast.Call,
         msg_override: str | None = None,
     ) -> None:
+        lineno = getattr(ast_node, "lineno", 1)
+        end_lineno = getattr(ast_node, "end_lineno", None)
+        complete_match = ast.get_source_segment(python_ast.content, ast_node)
+        if complete_match is None:
+            complete_match = get_complete_source_segment(lines, lineno, end_lineno)
+        start_column = getattr(ast_node, "col_offset", 0)
+        end_column = getattr(ast_node, "end_col_offset", start_column)
+        complete_identity = f"{complete_match}\x1f{start_column}:{end_column}"
         finding = AnalyzerFinding(
             rule_id=rule_id,
             message=msg_override or _RULE_MESSAGES[rule_id],
@@ -353,7 +360,8 @@ def _emit(
             confidence=_RULE_CONFIDENCES[rule_id],
             tags=[_TAG],
             context=get_context_from_lines(lines, lineno),
-            matched_text=get_source_segment(lines, lineno, end_lineno),
+            matched_text=complete_match[:200],
+            complete_match=complete_identity,
         )
         if budget is None:
             findings.append(finding)
@@ -374,9 +382,6 @@ def _emit(
         if call_name is None:
             continue
 
-        lineno = getattr(ast_node, "lineno", 1)
-        end_lineno = getattr(ast_node, "end_lineno", None)
-
         if call_name == "exec":
             if _is_chain_sink(ast_node, aliases) and ast_node.args:
                 source = _contains_dangerous_source(
@@ -385,8 +390,8 @@ def _emit(
                     budget.check_runtime if budget is not None else None,
                 )
                 if source:
-                    _emit("AST8", lineno, end_lineno, f"Dangerous chain: exec() wrapping {source}")
-            _emit("AST1", lineno, end_lineno)
+                    _emit("AST8", ast_node, f"Dangerous chain: exec() wrapping {source}")
+            _emit("AST1", ast_node)
 
         elif call_name == "eval":
             if _is_chain_sink(ast_node, aliases) and ast_node.args:
@@ -396,34 +401,34 @@ def _emit(
                     budget.check_runtime if budget is not None else None,
                 )
                 if source:
-                    _emit("AST8", lineno, end_lineno, f"Dangerous chain: eval() wrapping {source}")
-            _emit("AST2", lineno, end_lineno)
+                    _emit("AST8", ast_node, f"Dangerous chain: eval() wrapping {source}")
+            _emit("AST2", ast_node)
 
         elif call_name == "__import__":
-            _emit("AST3", lineno, end_lineno)
+            _emit("AST3", ast_node)
 
         elif call_name == "compile":
-            _emit("AST6", lineno, end_lineno)
+            _emit("AST6", ast_node)
 
         elif call_name.startswith("subprocess."):
             attr = call_name.split(".", 1)[1]
             if attr in _SUBPROCESS_CALLS:
-                _emit("AST4", lineno, end_lineno)
+                _emit("AST4", ast_node)
 
         elif call_name.startswith("os."):
             attr = call_name.split(".", 1)[1]
             if attr in _OS_EXEC_CALLS:
-                _emit("AST5", lineno, end_lineno)
+                _emit("AST5", ast_node)
 
         elif (deser_msg := _deserialization_message(call_name, ast_node)) is not None:
-            _emit("AST10", lineno, end_lineno, deser_msg)
+            _emit("AST10", ast_node, deser_msg)
 
         elif call_name == "getattr" and len(ast_node.args) >= 2:
             second_arg = ast_node.args[1]
             if not isinstance(second_arg, ast.Constant):
-                _emit("AST7", lineno, end_lineno)
+                _emit("AST7", ast_node)
             elif isinstance(second_arg.value, str) and second_arg.value in _DANGEROUS_GETATTR_NAMES:
-                _emit("AST9", lineno, end_lineno)
+                _emit("AST9", ast_node)
 
     return findings if budget is None else list(budget.current_findings)
 

src/skillspector/nodes/analyzers/behavioral_taint_tracking.py

@@ -48,8 +48,8 @@
 from .common import (
     apply_import_aliases,
     build_type_map,
+    get_complete_source_segment,
     get_context_from_lines,
-    get_source_segment,
     resolve_call_name_typed,
     resolve_dotted_name,
     resolve_dynamic_import_call,
@@ -477,6 +477,7 @@ def _emit(
         if key in seen:
             return
         seen.add(key)
+        complete_match = get_complete_source_segment(lines, lineno, end_lineno)
         finding = AnalyzerFinding(
             rule_id=rule_id,
             message=msg,
@@ -485,7 +486,8 @@ def _emit(
             confidence=_RULE_CONFIDENCES[rule_id],
             tags=[_TAG],
             context=get_context_from_lines(lines, lineno),
-            matched_text=get_source_segment(lines, lineno, end_lineno),
+            matched_text=complete_match[:200],
+            complete_match=complete_match,
         )
         if budget is None:
             findings.append(finding)

src/skillspector/nodes/analyzers/common.py

@@ -23,6 +23,8 @@
 from skillspector.models import Finding
 from skillspector.python_ast import build_import_aliases
 
+MAX_FINDING_CONTEXT_CHARS = 1_000
+
 
 def make_dummy_finding(analyzer_id: str) -> Finding:
     """Create a deterministic dummy finding for a stub analyzer."""
@@ -82,14 +84,38 @@ def get_context(content: str, match_start: int, context_lines: int = 3) -> str:
     match_line = content[:match_start].count("\n")
     start_line = max(0, match_line - context_lines)
     end_line = min(len(lines), match_line + context_lines + 1)
-    return "\n".join(lines[start_line:end_line])
+    selected_lines = lines[start_line:end_line]
+    if not selected_lines:
+        return ""
+    relative_line = min(match_line - start_line, len(selected_lines) - 1)
+    line_start = content.rfind("\n", 0, match_start) + 1
+    column = min(max(0, match_start - line_start), len(selected_lines[relative_line]))
+    anchor = sum(len(line) + 1 for line in selected_lines[:relative_line]) + column
+    return _bounded_context("\n".join(selected_lines), anchor)
 
 
 def get_context_from_lines(lines: list[str], lineno: int, window: int = 3) -> str:
     """Extract surrounding lines given pre-split *lines* and a 1-based *lineno*."""
     start = max(0, lineno - 1 - window)
     end = min(len(lines), lineno + window)
-    return "\n".join(lines[start:end])
+    selected_lines = lines[start:end]
+    if not selected_lines:
+        return ""
+    relative_line = min(max(0, lineno - 1 - start), len(selected_lines) - 1)
+    anchor = sum(len(line) + 1 for line in selected_lines[:relative_line])
+    return _bounded_context("\n".join(selected_lines), anchor)
+
+
+def _bounded_context(context: str, anchor: int) -> str:
+    """Return a bounded context window that retains the finding anchor."""
+    if len(context) <= MAX_FINDING_CONTEXT_CHARS:
+        return context
+    half_window = MAX_FINDING_CONTEXT_CHARS // 2
+    start = min(
+        max(0, anchor - half_window),
+        len(context) - MAX_FINDING_CONTEXT_CHARS,
+    )
+    return context[start : start + MAX_FINDING_CONTEXT_CHARS]
 
 
 def resolve_dotted_name(node: ast.expr) -> str | None:
@@ -287,8 +313,13 @@ def resolve_call_name_typed(
     return plain
 
 
-def get_source_segment(lines: list[str], lineno: int, end_lineno: int | None) -> str:
-    """Extract the source text for a given line range, truncated to 200 chars."""
+def get_complete_source_segment(lines: list[str], lineno: int, end_lineno: int | None) -> str:
+    """Extract the complete source text for a given line range."""
     start = max(0, lineno - 1)
     end = end_lineno or lineno
-    return "\n".join(lines[start:end])[:200]
+    return "\n".join(lines[start:end])
+
+
+def get_source_segment(lines: list[str], lineno: int, end_lineno: int | None) -> str:
+    """Extract a 200-character source preview for a given line range."""
+    return get_complete_source_segment(lines, lineno, end_lineno)[:200]

src/skillspector/nodes/analyzers/mcp_rug_pull.py

@@ -35,7 +35,7 @@
     ledger_event,
 )
 from skillspector.logging_config import get_logger
-from skillspector.models import Finding
+from skillspector.models import Finding, compute_match_fingerprint
 from skillspector.state import (
     AnalyzerNodeResponse,
     SkillspectorState,
@@ -238,6 +238,7 @@ def _check_rp1(
                     category=_CATEGORY,
                     tags=list(_TAGS),
                     matched_text=full_match[:200],
+                    match_fingerprint=compute_match_fingerprint("RP1", full_match),
                     explanation=(
                         "npx commands without a version suffix (e.g. @1.0.0) "
                         "create a rug-pull risk if the upstream server is "
@@ -272,6 +273,7 @@ def _check_rp1(
                     category=_CATEGORY,
                     tags=list(_TAGS),
                     matched_text=full_match[:200],
+                    match_fingerprint=compute_match_fingerprint("RP1", full_match),
                     explanation=(
                         "uvx/uv tool run commands without ==version create a rug-pull risk."
                     ),
@@ -307,6 +309,7 @@ def _check_rp1(
                     category=_CATEGORY,
                     tags=list(_TAGS),
                     matched_text=full_match[:200],
+                    match_fingerprint=compute_match_fingerprint("RP1", full_match),
                     explanation=(
                         "pip install without ==version installs the latest "
                         "release, which could include malicious changes."
@@ -333,6 +336,7 @@ def _check_rp1(
                     category=_CATEGORY,
                     tags=list(_TAGS),
                     matched_text=full_match[:200],
+                    match_fingerprint=compute_match_fingerprint("RP1", full_match),
                     explanation=(
                         "Docker image references without a specific tag (:latest "
                         "is implicit) or digest (@sha256:...) can be silently "
@@ -367,6 +371,7 @@ def _check_rp1(
                 category=_CATEGORY,
                 tags=list(_TAGS),
                 matched_text=m.group(0)[:200],
+                match_fingerprint=compute_match_fingerprint("RP1", m.group(0)),
                 explanation=(
                     "MCP server references in the skill manifest without version "
                     "pinning are a rug-pull risk."
@@ -399,6 +404,7 @@ def _check_rp2(manifest: dict, budget: _RugPullBudget) -> None:
                     category=_CATEGORY,
                     tags=list(_TAGS),
                     matched_text=m.group(0)[:200],
+                    match_fingerprint=compute_match_fingerprint("RP2", m.group(0)),
                     explanation=(
                         "Language in the manifest suggests the skill may request "
                         "additional permissions or tools in future versions. This "
@@ -425,18 +431,20 @@ def _check_rp3(manifest: dict, budget: _RugPullBudget) -> None:
         return
 
     version_str = str(version_value).strip()
+    version_preview = version_str[:200]
     if version_str in ("*", "latest", "any"):
         budget.emit(
             Finding(
                 rule_id="RP3",
-                message=f"Skill version is unpinned: '{version_str}'.",
+                message=f"Skill version is unpinned: '{version_preview}'.",
                 severity="LOW",
                 confidence=0.80,
                 file="SKILL.md",
                 start_line=1,
                 category=_CATEGORY,
                 tags=list(_TAGS),
-                matched_text=version_str,
+                matched_text=version_preview,
+                match_fingerprint=compute_match_fingerprint("RP3", version_str),
                 explanation=(
                     "An unpinned version allows automatic updates to any "
                     "future version, creating a rug-pull risk."
@@ -448,14 +456,15 @@ def _check_rp3(manifest: dict, budget: _RugPullBudget) -> None:
         budget.emit(
             Finding(
                 rule_id="RP3",
-                message=f"Skill version constraint may be too broad: '{version_str}'.",
+                message=f"Skill version constraint may be too broad: '{version_preview}'.",
                 severity="LOW",
                 confidence=0.40 if version_str.startswith(">=") else 0.50,
                 file="SKILL.md",
                 start_line=1,
                 category=_CATEGORY,
                 tags=list(_TAGS),
-                matched_text=version_str,
+                matched_text=version_preview,
+                match_fingerprint=compute_match_fingerprint("RP3", version_str),
                 explanation=(
                     "Broad version constraints allow automatic major-version "
                     "updates, which could silently introduce malicious changes."