pyreach

Documentation

Overview

Package pyreach implements a Tier-3 (package-level) reachability analyzer for Python packages. It walks application source files rooted at a Python project (pyproject.toml / setup.py / requirements.txt / Pipfile / poetry.lock / pdm.lock / uv.lock), scans every reachable .py file for top-level import statements, maps the imported module names to PyPI distribution names, and reports each registry vulnerability as Reachable / Unreachable / Unknown depending on whether the distribution appears in the import set (expanded transitively through the dep graph).

Tier-3 caveat: "unreachable" here means "the application source does not import this package, neither directly nor indirectly through app code". It does NOT mean "the vulnerability cannot be triggered" — Python is highly dynamic (importlib.import_module on user input, plugin discovery via entry points, Django INSTALLED_APPS strings, __import__ inside conditional branches). See docs/REACHABILITY.md for the full set of caveats.

The runner reads source in-process so users never need a Python interpreter or third-party tool on PATH. The Runner interface is preserved (rather than calling the scanner directly from the analyzer) so unit tests can inject a fake runner for deterministic behavior.

Index

• Constants
• type Analyzer
  • func (a Analyzer) Analyze(ctx context.Context, req model.AnalyzeRequest) (model.AnalyzeResult, error)
  • func (a Analyzer) Applicable(_ context.Context, req model.AnalyzeRequest) (bool, error)
  • func (a Analyzer) Descriptor() model.AnalyzerDescriptor
  • func (a Analyzer) Ready(context.Context, model.AnalyzeRequest) error
• type Runner
  • func NewRunner(logger *zap.Logger) Runner
• type RunnerResult

Constants

const Name = "pyreach"

Name is the analyzer's stable identifier (used in selectors and output).

Types

type Analyzer

type Analyzer struct {
	// Runner is the underlying scanner. Defaults to
	// NewRunner(Logger) when nil.
	Runner Runner
	Logger *zap.Logger
	// CacheDir overrides the default per-project result cache
	// location. Empty means "use the OS user cache directory under
	// bomly/analyzers/pyreach".
	CacheDir string
	// CacheTTL overrides the default 24h cache lifetime. Zero means
	// use the default.
	CacheTTL time.Duration
	// DisableCache turns off the on-disk result cache entirely.
	// Useful in CI smoke runs where freshness matters more than
	// speed.
	DisableCache bool
}

Analyzer is a Tier-3 (package-level) reachability analyzer for Python packages. It groups Python packages in the input graph by project root, runs the configured Runner once per project, and annotates each registry vulnerability on Python packages with a Reachability result.

Tier-3 caveat: "unreachable" here means "the application source does not import this distribution, neither directly nor indirectly through app code". It does NOT mean "the vulnerability cannot be triggered". Python is highly dynamic — importlib.import_module on user input, plugin discovery via entry points, Django INSTALLED_APPS strings — and none of those are visible to a static scanner. docs/REACHABILITY.md is the authoritative source on this.

func (Analyzer) Analyze

func (a Analyzer) Analyze(ctx context.Context, req model.AnalyzeRequest) (model.AnalyzeResult, error)

Analyze runs the configured Runner per discovered Python project root and writes Reachability onto every Python registry vulnerability in the graph. Errors degrade to Status=Unknown with a stable Reason — the engine relies on this to keep the pipeline running.

func (Analyzer) Applicable

func (a Analyzer) Applicable(_ context.Context, req model.AnalyzeRequest) (bool, error)

Applicable reports whether the request graph contains at least one Python package with attached vulnerabilities.

func (Analyzer) Descriptor

func (a Analyzer) Descriptor() model.AnalyzerDescriptor

Descriptor returns the registration metadata for the pyreach analyzer.

func (Analyzer) Ready

func (a Analyzer) Ready(context.Context, model.AnalyzeRequest) error

Ready reports whether the analyzer is callable. Always true; the runner surfaces missing-source / parse errors at Run time as Status=Unknown rather than blocking applicability.

type Runner

type Runner interface {
	// Name returns a stable identifier (e.g. "library") used in
	// telemetry and Reason fields.
	Name() string
	// Version returns the runner schema version. The result cache
	// folds it into its key so scanner upgrades invalidate prior
	// entries automatically.
	Version() string
	// Run walks projectDir and returns the imported-distribution set.
	// projectDir must contain at least one of pyproject.toml,
	// setup.py, setup.cfg, requirements*.txt, Pipfile, poetry.lock,
	// pdm.lock, or uv.lock.
	Run(ctx context.Context, projectDir string) (RunnerResult, error)
}

Runner walks a Python project rooted at projectDir and returns the set of PyPI distribution names imported anywhere in its reachable source tree. Implementations must NEVER panic and should map missing inputs, parse errors, and other recoverable conditions to a (RunnerResult, error) pair where the error is descriptive but does not abort the pipeline.

func NewRunner

func NewRunner(logger *zap.Logger) Runner

NewRunner returns the analyzer's Runner implementation, an in-process scanner that walks the project's .py source files and records every top-level module imported. Module names are mapped to PyPI distribution names through moduleToDistribution; the resulting set is what downstream BFS through the dep graph expands into the full reachable set.

type RunnerResult

type RunnerResult struct {
	// ImportedDistributions is the set of PyPI distribution names
	// (lowercase, hyphenated form — the canonical form of
	// distribution names) imported anywhere in the project's
	// reachable source tree. Module names are normalized to
	// distribution names through a layered strategy: a static map
	// of well-known mismatches (e.g. yaml -> PyYAML), then the
	// identity normalization (lowercase, "_" -> "-").
	ImportedDistributions map[string]struct{}
	// SourceFiles is the count of project .py files the runner
	// visited (for telemetry).
	SourceFiles int
	// SkippedDirs lists the directory names skipped during the walk
	// (venv/, __pycache__/, etc.) for debug logging.
	SkippedDirs []string
	// DynamicImportsDetected is true when the runner observed Python
	// dynamic-import constructs the static scanner cannot follow:
	// importlib.import_module(name), __import__(name) on a variable,
	// pkgutil.iter_modules / pkg_resources iterators. When true, an
	// "unreachable" verdict from this analyzer is necessarily
	// incomplete and the per-vuln Reachability.DynamicImportsDetected
	// flag is set.
	DynamicImportsDetected bool
}

RunnerResult is the parsed output of one runner pass over a project. It carries enough information for the analyzer to map advisories to reachable / unreachable / unknown without re-reading any source.