feat(python): classify executable source surfaces

src/skillspector/artifacts.py

@@ -99,6 +99,8 @@ class SecurityTextView:
     text: str
     source_offsets: array[int] | None = None
     reconstructions: tuple[SecurityTextReconstruction, ...] = ()
+    right_boundary_is_fixed: bool = False
+    right_boundary_recovery_start: int | None = None
 
     def source_offset(self, derived_offset: int) -> int:
         """Map a derived character offset to the corresponding source offset."""
@@ -228,6 +230,7 @@ class _ObfuscatedIgnoreState:
         ".markdown",
         ".txt",
         ".py",
+        ".pyw",
         ".sh",
         ".json",
         ".yaml",
@@ -401,6 +404,20 @@ def classify_artifact(path: str, data: bytes, *, referenced: bool = False) -> Ar
     }
 
 
+def promote_artifact_to_decoded_text(artifact: ArtifactRecord) -> None:
+    """Apply a successful format-aware text decode without erasing prior limits."""
+    generic_binary_scope = artifact["content_kind"] is ContentKind.BINARY and (
+        artifact["disposition"] is ArtifactDisposition.OUT_OF_SCOPE
+        or artifact["disposition"] is ArtifactDisposition.PARTIAL
+        and "reason" not in artifact
+    )
+    artifact["content_kind"] = ContentKind.TEXT
+    artifact["decodable"] = True
+    artifact["misleading_extension"] = _suffix(artifact["path"]) in _BINARY_EXTENSIONS
+    if generic_binary_scope:
+        artifact["disposition"] = ArtifactDisposition.ANALYZED
+
+
 def decode_text(data: bytes) -> str:
     """Return the loss-tolerant local text projection for static analyzers."""
     return data.decode("utf-8", errors="replace")
@@ -1679,6 +1696,30 @@ def _contextual_default_ignorable_spans(
                 yield span_start, end
 
 
+def _normalization_ignored_spans_in_gap(
+    text: str,
+    gap_start: int,
+    gap_end: int,
+) -> Iterator[tuple[int, int]]:
+    """Yield the normalized-view removals inside one contextual gap."""
+    for match in _DEFAULT_IGNORABLE_RUN_PATTERN.finditer(text, gap_start, gap_end):
+        start, end = match.span()
+        ignored_start = (
+            start
+            if _is_unconditionally_ignored(text[start])
+            or _is_contextual_default_ignorable_offset(text, start)
+            else start + 1
+        )
+        ignored_end = (
+            end
+            if _is_unconditionally_ignored(text[end - 1])
+            or _is_contextual_default_ignorable_offset(text, end - 1)
+            else end - 1
+        )
+        if ignored_start < ignored_end:
+            yield ignored_start, ignored_end
+
+
 def _normalization_ignored_spans(
     text: str,
     check_runtime: Callable[[], None] | None = None,
@@ -1689,22 +1730,7 @@ def _normalization_ignored_spans(
         require_word_boundaries=False,
         check_runtime=check_runtime,
     ):
-        for match in _DEFAULT_IGNORABLE_RUN_PATTERN.finditer(text, gap_start, gap_end):
-            start, end = match.span()
-            ignored_start = (
-                start
-                if _is_unconditionally_ignored(text[start])
-                or _is_contextual_default_ignorable_offset(text, start)
-                else start + 1
-            )
-            ignored_end = (
-                end
-                if _is_unconditionally_ignored(text[end - 1])
-                or _is_contextual_default_ignorable_offset(text, end - 1)
-                else end - 1
-            )
-            if ignored_start < ignored_end:
-                yield ignored_start, ignored_end
+        yield from _normalization_ignored_spans_in_gap(text, gap_start, gap_end)
 
 
 def _contextual_default_ignorable_offsets(
@@ -1813,6 +1839,55 @@ def normalized_security_view(
     return SecurityTextView("normalized", output.getvalue(), offsets)
 
 
+def normalized_security_prefix(text: str, max_chars: int) -> str:
+    """Return an exact bounded prefix of the normalized security projection."""
+    if max_chars <= 0:
+        return ""
+
+    output = StringIO()
+    output_chars = 0
+
+    def append(character: str) -> bool:
+        nonlocal output_chars
+        normalized = unicodedata.normalize("NFKC", character).translate(ASCII_CONFUSABLE_SKELETON)
+        remaining = max_chars - output_chars
+        output.write(normalized[:remaining])
+        output_chars += min(len(normalized), remaining)
+        return output_chars >= max_chars
+
+    source_offset = 0
+    while source_offset < len(text) and output_chars < max_chars:
+        if not _is_token_gap_character(text[source_offset]):
+            if append(text[source_offset]):
+                break
+            source_offset += 1
+            continue
+
+        gap_start = source_offset
+        while source_offset < len(text) and _is_token_gap_character(text[source_offset]):
+            source_offset += 1
+        gap_end = source_offset
+        before_is_word = gap_start > 0 and _is_word_character(text[gap_start - 1])
+        after_is_word = gap_end < len(text) and _is_word_character(text[gap_end])
+        ignored_spans = iter(
+            _normalization_ignored_spans_in_gap(text, gap_start, gap_end)
+            if before_is_word or after_is_word
+            else ()
+        )
+        next_ignored = next(ignored_spans, None)
+        gap_offset = gap_start
+        while gap_offset < gap_end and output_chars < max_chars:
+            if next_ignored is not None and gap_offset == next_ignored[0]:
+                gap_offset = next_ignored[1]
+                next_ignored = next(ignored_spans, None)
+                continue
+            if not _is_unconditionally_ignored(text[gap_offset]) and append(text[gap_offset]):
+                break
+            gap_offset += 1
+
+    return output.getvalue()
+
+
 def obfuscated_instruction_view(
     text: str,
     check_runtime: Callable[[], None] | None = None,
@@ -2171,6 +2246,35 @@ def _requires_normalized_security_view_uncached(
     return not text.translate(_REMOVE_ALLOWED_FORMAT_CHARACTERS).isprintable()
 
 
+def _has_derived_security_view(text: str) -> bool:
+    """Return whether security projection produces a distinct text view."""
+    if text.isascii():
+        return (
+            _IGNORED_ASCII_CONTROL.search(text) is not None
+            or _has_letter_spacing_run(text)
+            or next(_obfuscated_instruction_matches(text), None) is not None
+        )
+    if any(
+        unicodedata.normalize("NFKC", character).translate(ASCII_CONFUSABLE_SKELETON) != character
+        for character in text
+    ):
+        return True
+    if any(_is_unconditionally_ignored(character) for character in text):
+        return True
+    if next(_normalization_ignored_spans(text), None) is not None:
+        return True
+    if "\ufffd" in text or next(_compact_gap_offsets(text), None) is not None:
+        return True
+    return (
+        _has_letter_spacing_run(text)
+        or next(
+            _obfuscated_instruction_matches(text),
+            None,
+        )
+        is not None
+    )
+
+
 def security_text_views(
     text: str,
     check_runtime: Callable[[], None] | None = None,

src/skillspector/cli.py

@@ -57,7 +57,7 @@
 )
 from skillspector.logging_config import get_logger, set_level
 from skillspector.mcp_registry import scan_registry
-from skillspector.models import Finding
+from skillspector.models import OCCURRENCE_FINDING_ID_KEY, Finding
 from skillspector.multi_skill import MultiSkillDetectionResult, SkillDirectory, detect_skills
 from skillspector.nodes.analyzers import ANALYZER_MODULES, ANALYZER_NODE_IDS
 from skillspector.nodes.report import report
@@ -1164,18 +1164,32 @@ def _cache_transitive_result(
     child_filtered = _coerce_findings_list(child_result.get("filtered_findings"))
     child_findings = _coerce_findings_list(child_result.get("findings"))
     all_ids = {finding.finding_id for finding in [*child_filtered, *child_findings]}
+    all_ids.update(
+        occurrence_id
+        for finding in [*child_filtered, *child_findings]
+        for occurrence in finding.occurrences
+        if isinstance((occurrence_id := occurrence.get(OCCURRENCE_FINDING_ID_KEY)), str)
+    )
     all_ids.update(_effective_finding_ids(child_result))
     finding_id_map = {
         finding_id: _scoped_finding_id(source_identity, finding_id) for finding_id in all_ids
     }
 
     def _scope_finding(finding: Finding) -> Finding:
+        occurrences = []
+        for raw in finding.occurrences:
+            occurrence = dict(raw)
+            occurrence_id = occurrence.get(OCCURRENCE_FINDING_ID_KEY)
+            if isinstance(occurrence_id, str):
+                occurrence[OCCURRENCE_FINDING_ID_KEY] = finding_id_map[occurrence_id]
+            occurrences.append(occurrence)
         return replace(
             finding,
             finding_id=finding_id_map[finding.finding_id],
             source_url=target,
             source_identity=source_identity,
             source_digest=source_digest,
+            occurrences=occurrences,
         )
 
     scoped_filtered = [_scope_finding(item) for item in child_filtered[:_TRANSITIVE_MAX_FINDINGS]]

src/skillspector/input_handler.py

@@ -81,6 +81,7 @@
 _DIRECT_FILE_URL_SUFFIXES = (
     ".md",
     ".py",
+    ".pyw",
     ".sh",
 )
 

src/skillspector/inspection_ledger.py

@@ -98,6 +98,8 @@ class LedgerReason(StrEnum):
     OUTPUT_LIMIT = "output_limit"
     TRANSITIVE_CHILD_SCAN_FAILED = "transitive_child_scan_failed"
     STATIC_PARSE_LIMIT = "static_parse_limit"
+    PYTHON_SOURCE_AMBIGUOUS = "python_source_ambiguous"
+    PYTHON_SOURCE_DECODE_ERROR = "python_source_decode_error"
     OBFUSCATED_INSTRUCTION_TEXT = "obfuscated_instruction_text"
 
 
@@ -210,6 +212,12 @@ class LedgerReason(StrEnum):
     LedgerReason.STATIC_PARSE_LIMIT: (
         "A security-relevant expression exceeded a bounded static parser's span limit."
     ),
+    LedgerReason.PYTHON_SOURCE_AMBIGUOUS: (
+        "Python execution intent depends on runtime or platform-specific shebang semantics."
+    ),
+    LedgerReason.PYTHON_SOURCE_DECODE_ERROR: (
+        "Python source bytes could not be decoded under their declared encoding."
+    ),
     LedgerReason.OBFUSCATED_INSTRUCTION_TEXT: (
         "Obfuscated instruction text could not be fully evaluated by the deterministic layer."
     ),

src/skillspector/models.py

@@ -120,6 +120,11 @@ def _new_finding_id() -> str:
     return f"finding-{uuid4().hex}"
 
 
+OCCURRENCE_FINDING_ID_KEY = "_skillspector_report_finding_id"
+OCCURRENCE_CODE_SNIPPET_KEY = "_skillspector_report_code_snippet"
+_PRIVATE_OCCURRENCE_KEYS = frozenset({OCCURRENCE_FINDING_ID_KEY, OCCURRENCE_CODE_SNIPPET_KEY})
+
+
 @dataclass
 class Finding:
     """Finding model for graph state and report output (shape aligned with to_dict)."""
@@ -218,7 +223,9 @@ def _serialized_occurrences(self) -> list[dict[str, object]]:
         ]
         serialized: list[dict[str, object]] = []
         for raw in occurrences:
-            occurrence = dict(raw)
+            occurrence = {
+                key: value for key, value in raw.items() if key not in _PRIVATE_OCCURRENCE_KEYS
+            }
             if self.source_identity:
                 occurrence.setdefault("source_identity", self.source_identity)
             if self.source_digest:

src/skillspector/nested_artifacts.py

@@ -40,6 +40,7 @@
     LedgerRecordType,
     ledger_event,
 )
+from skillspector.python_ast import PythonSourceClassification, classify_python_source
 
 ARCHIVE_MAX_DEPTH = 3
 ARCHIVE_MAX_MEMBERS = 1_000
@@ -79,6 +80,7 @@
         ".phtml",
         ".ps1",
         ".py",
+        ".pyw",
         ".pyc",
         ".pyo",
         ".rb",
@@ -113,6 +115,11 @@ class NestedInspectionResult:
     components: list[str] = field(default_factory=list)
     file_cache: dict[str, str] = field(default_factory=dict)
     raw_file_cache: dict[str, bytes] = field(default_factory=dict)
+    # Classify with the archive member's execution path, while retaining the
+    # virtual path as the stable cache/report key used by downstream analyzers.
+    python_source_classifications: dict[str, PythonSourceClassification] = field(
+        default_factory=dict
+    )
     artifact_inventory: list[ArtifactRecord] = field(default_factory=list)
     metadata: list[dict[str, object]] = field(default_factory=list)
     outer_metadata: dict[str, dict[str, object]] = field(default_factory=dict)
@@ -435,14 +442,22 @@ def _record_outer_metadata(
     }
 
 
-def _virtual_type(path: str, data: bytes, nested_type: str | None) -> str:
+def _virtual_type(
+    path: str,
+    data: bytes,
+    nested_type: str | None,
+    source_classification: PythonSourceClassification,
+) -> str:
     if nested_type is not None:
         return nested_type
+    if source_classification is PythonSourceClassification.PYTHON:
+        return "python"
     suffix = Path(path).suffix.lower()
     return {
         ".md": "markdown",
         ".markdown": "markdown",
         ".py": "python",
+        ".pyw": "python",
         ".sh": "shell",
         ".bash": "shell",
         ".zsh": "shell",
@@ -994,10 +1009,17 @@ def _inspect_zip_bytes(
             executable = _member_executable(info, safe_name, member_data)
             member_hidden = _is_hidden_path(safe_name)
             concealed = executable and bool(concealment_reasons)
-            virtual_type = _virtual_type(safe_name, member_data, nested_type)
+            source_classification = classify_python_source(safe_name, member_data)
+            virtual_type = _virtual_type(
+                safe_name,
+                member_data,
+                nested_type,
+                source_classification,
+            )
             result.components.append(virtual_path)
             result.file_cache[virtual_path] = member_data.decode("utf-8", errors="replace")
             result.raw_file_cache[virtual_path] = member_data
+            result.python_source_classifications[virtual_path] = source_classification
             artifact = classify_artifact(virtual_path, member_data)
             result.artifact_inventory.append(artifact)
             result.metadata.append(