source/pkg/bundle/analyzer_python.go

package bundle

import "strings"

// PythonSourceAnalyzer inspects .py source via an AST-free textual scan. It
// reuses the shared source/sink vocabulary (os.environ/os.getenv harvesting,
// subprocess/os.system, socket/requests/urllib exfil, eval/exec of decoded
// payloads, pip/registry rewrites). SevHigh+Corroborated only when an
// env/secret source co-occurs with a network sink; an isolated indicator is
// SevMedium. A bare `import os` / `import requests` with no source+sink pairing
// emits nothing.
type PythonSourceAnalyzer struct{}

func (PythonSourceAnalyzer) Name() string { return "python-source" }

func (PythonSourceAnalyzer) Handles(kind FileKind) bool { return kind == KindPythonSource }

func (PythonSourceAnalyzer) Analyze(f *File, b *Bundle) ([]Finding, error) {
	if f == nil {
		return nil, nil
	}
	text := string(f.Sniff)
	out := sharedIndicatorScan(text, f.RelPath, "python-source")

	// Python-specific: eval/exec of decoded content is RCE even without a pipe.
	lines := strings.Split(text, "\n")
	for i, l := range lines {
		low := strings.ToLower(l)
		if (strings.Contains(low, "eval(") || strings.Contains(low, "exec(")) &&
			(strings.Contains(low, "b64decode") || strings.Contains(low, "base64") ||
				strings.Contains(low, "decompress") || strings.Contains(low, "fromhex")) {
			out = append(out, Finding{
				Analyzer:     "python-source",
				File:         f.RelPath,
				Signal:       "eval-decoded-payload",
				Severity:     SevHigh,
				Detail:       "eval/exec of a decoded/decompressed payload",
				Line:         i + 1,
				Corroborated: true,
			})
		}
	}

	if pe, ok := paddingEvasionFinding(f, "python-source"); ok {
		out = append(out, pe)
	}
	return dedupeFindings(out), nil
}