source/pkg/bundle/analyzer_pyc.go

package bundle

import (
	"path/filepath"
	"strings"
)

// PycAnalyzer inspects compiled Python bytecode (.pyc) — the xz-utils "shipped
// artifact != source" vector. ANY shipped .pyc is SevHigh-by-default. It parses
// the pyc header and marshal-decodes the top-level code object to recover
// co_names/co_consts/strings, scans them for exfil primitives, and diffs the
// recovered symbols against the same-stem sibling .py so a clean decoy source
// cannot launder a malicious .pyc. Decoding is PURE-GO ONLY: the scanner never
// shells out to the system python (`python3 -m dis`) on an untrusted .pyc —
// disassembling attacker-supplied bytecode through the interpreter is a
// code-execution surface and the exact payload class this tool exists to flag.
// Bytecode that pure-Go cannot decode is reported as a SevHigh-Opaque artifact,
// never executed. Never panics on truncated/forged-magic/malformed bytecode.
type PycAnalyzer struct{}

func (PycAnalyzer) Name() string { return "pyc" }

func (PycAnalyzer) Handles(kind FileKind) bool { return kind == KindPyc }

func (PycAnalyzer) Analyze(f *File, b *Bundle) ([]Finding, error) {
	if f == nil {
		return nil, nil
	}
	out := []Finding{
		{
			Analyzer:   "pyc",
			File:       f.RelPath,
			Signal:     "ships-compiled-bytecode",
			Severity:   SevHigh,
			Detail:     "ships compiled Python bytecode (shipped artifact may differ from source)",
			Structural: true,
		},
	}

	// Pure-Go decode only. We deliberately do NOT fall back to `python3 -m dis`
	// (or any subprocess/interpreter) on an untrusted .pyc: running the system
	// python against attacker-supplied bytecode is a code-execution surface.
	// Undecodable bytecode is an opaque artifact (which escalates), not a reason
	// to execute it.
	symbols, ok := recoverPycStrings(f.Sniff)
	if !ok {
		out = append(out, Finding{
			Analyzer:   "pyc",
			File:       f.RelPath,
			Signal:     "opaque-bytecode",
			Severity:   SevHigh,
			Detail:     "compiled bytecode could not be decoded (opaque artifact)",
			Opaque:     true,
			Structural: true,
		})
		return out, nil
	}

	// Scan recovered strings for exfil/credential/network primitives.
	joined := strings.Join(symbols, "\n")
	out = append(out, sharedIndicatorScan(joined, f.RelPath, "pyc")...)

	// compiled-source-mismatch: symbols present in bytecode but absent from the
	// same-stem sibling source. A clean decoy .py does NOT satisfy
	// "compiled-without-matching-source".
	if src, found := siblingSourceFor(b, f); found {
		srcText := strings.ToLower(string(src.Sniff))
		var missing []string
		for _, sym := range symbols {
			if !isInterestingSymbol(sym) {
				continue
			}
			if !strings.Contains(srcText, strings.ToLower(sym)) {
				missing = append(missing, sym)
				if len(missing) >= 8 {
					break
				}
			}
		}
		if len(missing) > 0 {
			out = append(out, Finding{
				Analyzer:     "pyc",
				File:         f.RelPath,
				Signal:       "compiled-source-mismatch",
				Severity:     SevHigh,
				Detail:       "bytecode references symbols absent from sibling source: " + strings.Join(missing, ", "),
				Corroborated: true,
			})
		}
	} else {
		out = append(out, Finding{
			Analyzer: "pyc",
			File:     f.RelPath,
			Signal:   "compiled-without-matching-source",
			Severity: SevHigh,
			Detail:   "compiled bytecode ships with no same-stem source file (uninspectable, xz pattern)",
			// Escalating, host-agnostic: shipping executable Python BYTECODE with no
			// source in a distributed skill is the opaque-execution payload and is
			// not a benign idiom (verified: no benign corpus bundle ships a
			// sourceless .pyc). Unlike a native .so (which a legit skill may vendor),
			// sourceless .pyc has no legitimate distribution reason.
			Corroborated: true,
		})
	}

	return dedupeFindings(out), nil
}

// recoverPycStrings attempts a minimal pure-Go recovery of printable
// identifier/const strings from a CPython .pyc marshal stream. It does NOT fully
// implement the marshal format; instead it validates the pyc header, then walks
// the marshal body extracting length-prefixed string objects by their payloads.
// Best-effort and bounded; returns ok=false when the header is not a plausible
// pyc so the caller can fall back. Never panics.
func recoverPycStrings(data []byte) (symbols []string, ok bool) {
	defer func() {
		if recover() != nil {
			symbols = nil
			ok = false
		}
	}()

	// pyc header: 4-byte magic, then (3.7+) 4-byte bit field, 4-byte mtime,
	// 4-byte source size => 16-byte header. CPython magic ends with \r\n.
	if len(data) < 16 {
		return nil, false
	}
	if data[2] != 0x0d || data[3] != 0x0a {
		return nil, false
	}

	body := data[16:]
	syms := walkMarshalStrings(body)
	// Header was valid => decode succeeded even if zero strings were found.
	return syms, true
}

// marshal type codes for string-like objects (flag bit 0x80 = interned/ref).
const (
	marshalString     = 's' // TYPE_STRING (bytes), 4-byte length prefix
	marshalUnicode    = 'u' // TYPE_UNICODE, 4-byte length prefix
	marshalInterned   = 't' // TYPE_INTERNED, 4-byte length prefix
	marshalShortASCII = 'z' // TYPE_SHORT_ASCII, 1-byte length prefix
	marshalShortInt   = 'Z' // TYPE_SHORT_ASCII_INTERNED, 1-byte length prefix
	marshalASCII      = 'a' // TYPE_ASCII, 4-byte length prefix
	marshalASCIIInt   = 'A' // TYPE_ASCII_INTERNED, 4-byte length prefix
)

// walkMarshalStrings linearly scans the marshal body for string-typed objects
// and lifts their payloads. Intentionally a tolerant scanner (not a full
// recursive unmarshaller): it slides over the bytes, and whenever it sees a
// recognized string type code followed by a plausible length, it extracts the
// payload. Bounded by input length and a string cap; safe on truncated data.
func walkMarshalStrings(body []byte) []string {
	var out []string
	n := len(body)
	i := 0
	const maxStrings = 4096
	for i < n && len(out) < maxStrings {
		c := body[i] & 0x7f // strip the ref flag
		switch c {
		case marshalString, marshalUnicode, marshalInterned, marshalASCII, marshalASCIIInt:
			if i+5 > n {
				i++
				continue
			}
			length := int(body[i+1]) | int(body[i+2])<<8 | int(body[i+3])<<16 | int(body[i+4])<<24
			if length < 0 || length > 1<<16 || i+5+length > n {
				i++
				continue
			}
			s := string(body[i+5 : i+5+length])
			if isPrintableRun(s) {
				out = append(out, s)
				i += 5 + length
				continue
			}
			i++
		case marshalShortASCII, marshalShortInt:
			if i+2 > n {
				i++
				continue
			}
			length := int(body[i+1])
			if i+2+length > n {
				i++
				continue
			}
			s := string(body[i+2 : i+2+length])
			if isPrintableRun(s) {
				out = append(out, s)
				i += 2 + length
				continue
			}
			i++
		default:
			i++
		}
	}
	return dedupeStrings(out)
}

// siblingSourceFor returns the same-stem .py sibling of a .pyc, if present.
// Handles the CPython cache naming "name.cpython-312.pyc" -> "name.py".
func siblingSourceFor(b *Bundle, pyc *File) (*File, bool) {
	if b == nil {
		return nil, false
	}
	stem := pycStem(filepath.Base(pyc.RelPath))
	dir := filepath.Dir(pyc.RelPath)
	want := stem + ".py"
	for _, f := range b.Files {
		if f == nil || f == pyc {
			continue
		}
		if f.Kind != KindPythonSource {
			continue
		}
		if filepath.Dir(f.RelPath) != dir {
			continue
		}
		if filepath.Base(f.RelPath) == want {
			return f, true
		}
	}
	return nil, false
}

// pycStem strips ".pyc" and an optional ".cpython-XYZ"/".opt-N" cache tag.
func pycStem(base string) string {
	base = strings.TrimSuffix(base, ".pyc")
	if idx := strings.Index(base, ".cpython-"); idx >= 0 {
		base = base[:idx]
	}
	if idx := strings.Index(base, ".opt-"); idx >= 0 {
		base = base[:idx]
	}
	return base
}

// isInterestingSymbol filters recovered marshal strings down to plausible
// identifiers/dotted-attrs/exfil tokens worth diffing against source.
func isInterestingSymbol(s string) bool {
	s = strings.TrimSpace(s)
	if len(s) < 3 || len(s) > 200 {
		return false
	}
	if strings.ContainsAny(s, " \t") && !strings.Contains(s, "://") {
		return false
	}
	return true
}

func dedupeStrings(in []string) []string {
	seen := map[string]bool{}
	var out []string
	for _, s := range in {
		if s == "" || seen[s] {
			continue
		}
		seen[s] = true
		out = append(out, s)
	}
	return out
}

// isPrintableRun reports whether s is mostly printable ASCII (a recovered
// string, not random bytes).
func isPrintableRun(s string) bool {
	if s == "" {
		return false
	}
	runes := []rune(s)
	printable := 0
	for _, r := range runes {
		if r >= 0x20 && r < 0x7f {
			printable++
		}
	}
	return float64(printable)/float64(len(runes)) >= 0.85
}