feat(python): classify executable source surfaces

src/skillspector/artifacts.py

@@ -230,6 +230,7 @@ class _ObfuscatedIgnoreState:
         ".markdown",
         ".txt",
         ".py",
+        ".pyw",
         ".sh",
         ".json",
         ".yaml",
@@ -403,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")

src/skillspector/cli.py

@@ -59,7 +59,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,
@@ -1314,18 +1314,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

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

src/skillspector/inspection_ledger.py

@@ -108,6 +108,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"
 
 
@@ -221,6 +223,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

@@ -41,6 +41,7 @@
     LedgerRecordType,
     ledger_event,
 )
+from skillspector.python_ast import PythonSourceClassification, classify_python_source
 
 ARCHIVE_MAX_DEPTH = 3
 ARCHIVE_MAX_MEMBERS = 1_000
@@ -80,6 +81,7 @@
         ".phtml",
         ".ps1",
         ".py",
+        ".pyw",
         ".pyc",
         ".pyo",
         ".rb",
@@ -114,6 +116,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)
@@ -539,14 +546,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",
@@ -1098,10 +1113,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(

src/skillspector/nodes/analyzers/behavioral_ast.py

@@ -32,7 +32,12 @@
 )
 from skillspector.logging_config import get_logger
 from skillspector.models import AnalyzerFinding, Finding, Location, Severity
-from skillspector.python_ast import ParsedPythonFile, get_python_ast
+from skillspector.python_ast import (
+    ParsedPythonFile,
+    PythonSourceClassification,
+    get_python_ast,
+    resolve_python_source_classification,
+)
 from skillspector.state import (
     AnalyzerNodeResponse,
     SkillspectorState,
@@ -648,14 +653,42 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse:
     """Parse Python files via AST and detect dangerous execution patterns."""
     components: list[str] = state.get("components") or []
     file_cache: dict[str, str] = state.get("local_file_cache") or state.get("file_cache") or {}
+    raw_file_cache = state.get("raw_file_cache")
+    source_classifications = (
+        state.get("python_source_classifications")
+        if "python_source_classifications" in state
+        else None
+    )
+    source_classification_limitations = state.get("python_source_classification_limitations") or {}
+    source_decode_failures = state.get("python_source_decode_failures") or {}
     python_ast_cache_key = state.get("python_ast_cache_key")
     all_findings: list[Finding] = []
     ledger_events: list[InspectionLedgerEvent] = []
     budget = _BehavioralBudget(state)
     terminal_limit: _BehavioralResourceLimitError | None = None
 
     for path in components:
-        if not path.endswith(".py"):
+        content = file_cache.get(path)
+        source_classification: PythonSourceClassification | None = None
+        if source_classifications is not None and path in source_classifications:
+            source_classification = resolve_python_source_classification(
+                path,
+                content,
+                source_classifications=source_classifications,
+                raw_file_cache=raw_file_cache,
+            )
+            if source_classification is PythonSourceClassification.NON_PYTHON:
+                continue
+        if path in source_classification_limitations:
+            ledger_events.append(
+                ledger_event(
+                    outcome=LedgerOutcome.PARTIAL,
+                    phase="behavioral",
+                    analyzer_id=ANALYZER_ID,
+                    path=path,
+                    reason=LedgerReason.RUNTIME_LIMIT,
+                )
+            )
             continue
         if terminal_limit is None and budget.analyzer_exhausted():
             terminal_limit = _BehavioralResourceLimitError(
@@ -665,11 +698,41 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse:
                     "limit_findings": MAX_FINDINGS_PER_ANALYZER,
                 },
             )
+        if terminal_limit is None:
+            try:
+                budget.check_runtime()
+            except _BehavioralResourceLimitError as exc:
+                terminal_limit = exc
         if terminal_limit is not None:
             event = _partial_limit_event(path, terminal_limit)
             ledger_events.append(event)
             continue
-        content = file_cache.get(path)
+        if path in source_decode_failures:
+            ledger_events.append(
+                ledger_event(
+                    outcome=LedgerOutcome.PARTIAL,
+                    phase="behavioral",
+                    analyzer_id=ANALYZER_ID,
+                    path=path,
+                    reason=LedgerReason.PYTHON_SOURCE_DECODE_ERROR,
+                )
+            )
+            continue
+        if source_classification is None:
+            source_classification = resolve_python_source_classification(
+                path,
+                content,
+                source_classifications=source_classifications,
+                raw_file_cache=raw_file_cache,
+            )
+            try:
+                budget.check_runtime()
+            except _BehavioralResourceLimitError as exc:
+                terminal_limit = exc
+                ledger_events.append(_partial_limit_event(path, exc))
+                continue
+        if source_classification is PythonSourceClassification.NON_PYTHON:
+            continue
         if content is None:
             event = ledger_event(
                 outcome=LedgerOutcome.FAILED,
@@ -725,10 +788,19 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse:
                 )
             else:
                 event = ledger_event(
-                    outcome=LedgerOutcome.COMPLETED,
+                    outcome=(
+                        LedgerOutcome.PARTIAL
+                        if source_classification is PythonSourceClassification.AMBIGUOUS
+                        else LedgerOutcome.COMPLETED
+                    ),
                     phase="behavioral",
                     analyzer_id=ANALYZER_ID,
                     path=path,
+                    reason=(
+                        LedgerReason.PYTHON_SOURCE_AMBIGUOUS
+                        if source_classification is PythonSourceClassification.AMBIGUOUS
+                        else None
+                    ),
                     emitted_finding_ids=[finding.finding_id for finding in path_findings],
                 )
         ledger_events.append(event)