engine.py

"""Composition root: the full analysis DAG, plugin loading and one check run.

This is the only module that knows every layer. The CLI and future hosts call it;
plugins and analyses never import it.
"""

from __future__ import annotations

import concurrent.futures
import hashlib
import multiprocessing
import sys
import tempfile
from collections.abc import Callable, Iterable, Mapping, Sequence
from dataclasses import dataclass, field, replace
from pathlib import Path
from types import MappingProxyType
from typing import Any, ClassVar

from coretrace_python import __version__
from coretrace_python.abstract import ConstantPropagation, HeapAnalysis, RangeAnalysis
from coretrace_python.abstract.module_constants import ModuleCollectionsAnalysis
from coretrace_python.abstract.strings import ModuleStringsAnalysis
from coretrace_python.analysis import (
    AnalysisContext,
    AnalysisManager,
    AnyAnalysis,
    TransformationPass,
)
from coretrace_python.cache import (
    CachedModule,
    ProjectCache,
    decode,
    directory_fingerprint,
    encode,
    fingerprint,
    module_keys,
)
from coretrace_python.cfg import CFGAnalysis, CFGError, DominanceAnalysis, PostDominanceAnalysis
from coretrace_python.dependency import (
    ADVISORY_FILE,
    DEPENDENCY_FILES,
    POLICY_FILE,
    Advisory,
    AdvisoryFileError,
    DependencyAnalysis,
    DependencyGraph,
    Policy,
    apply_policy,
    load_advisories,
    load_policy,
    parse_dependencies,
)
from coretrace_python.dependency.correlation import (
    Affected,
    SuffixIndex,
    advisory_sinks,
    advisory_suffix_sinks,
    affected_symbols,
    correlate,
    suffix_index,
)
from coretrace_python.findings import (
    ADVISORIES,
    FINDING_SCHEMA_VERSION,
    PLUGIN,
    Component,
    Confidence,
    Coverage,
    FileCoverage,
    Finding,
    RuleDescription,
    Severity,
)
from coretrace_python.findings.context import code_context, code_origin, path_in_project
from coretrace_python.findings.refutation import RefutationAnalysis
from coretrace_python.findings.suppressions import partition
from coretrace_python.frontend import HIRBuildError, ParseError, build_hir
from coretrace_python.hir import HIR_SCHEMA_VERSION, nodes
from coretrace_python.interprocedural import (
    CallGraph,
    CallGraphAnalysis,
    CallSite,
    ClearingAnalysis,
    FunctionSummary,
    ModuleFunction,
    ModuleGraph,
    ProjectSummaries,
    SignaturesAnalysis,
    SummaryAnalysis,
    SummaryIndex,
    SymbolRead,
    TemplateCalls,
    TemplateCallsAnalysis,
    build_module_graph,
    discover_sources,
    import_names,
    name_modules,
    project_symbol,
    starting_summaries,
)
from coretrace_python.ir import PYIR_SCHEMA_VERSION
from coretrace_python.ir.defuse import DefUseAnalysis
from coretrace_python.ir.lowering import (
    LoweringError,
    ModuleIRAnalysis,
    PyIRAnalysis,
    analyzable_functions,
)
from coretrace_python.ir.ssa import SSAAnalysis
from coretrace_python.plugins import (
    PLUGIN_API_VERSION,
    Plugin,
    PluginRegistry,
    ProjectContext,
    ProjectPlugin,
    apply_refinement,
    discover_plugins,
    run_plugins,
)
from coretrace_python.reporters import Report
from coretrace_python.semantic import SEMANTIC_ANALYSES
from coretrace_python.semantic.imports import ImportAnalysis, ImportResolutionError, ImportTable
from coretrace_python.semantic.scopes import ScopeAnalysis, ScopeError
from coretrace_python.semantic.symbols import MembersAnalysis, SymbolAnalysis, SymbolId
from coretrace_python.source import SourceFile, SourceId, SourceManager, SourceSpan, decode_text
from coretrace_python.taint import (
    EntryPoint,
    EntryPointAnalysis,
    EscapedTemplates,
    ModelTable,
    RegisteredRoutes,
    Routes,
    SecurityModelAnalysis,
    SecurityModelRegistry,
    TaintAnalysis,
    TaintKind,
    escaped_templates,
    project_templates,
    registered_routes,
    request_processor,
    unread_renders,
)
from coretrace_python.taint.urls import flow_url

TOOL_NAME = "coretrace-python-analyzer"

# The plugins shipped with the package: models, detectors, secrets and dependency
# checks. They are loaded by default; ``--plugins`` adds directories on top.
BUNDLED_PLUGINS = Path(__file__).resolve().parent / "bundled"


# The rules the engine reports itself, beside those of the plugins: notes on what it could not
# analyse, and the correlation of a dependency's advisory with the project's taint flows.
ENGINE_RULES: Mapping[str, RuleDescription] = MappingProxyType(
    {
        "syntax-error": RuleDescription(
            "File that could not be read",
            "A source, requirements or policy file the analyzer could not read or parse: "
            "nothing in it is checked.",
        ),
        "ambiguous-module": RuleDescription(
            "Ambiguous module name",
            "A file whose import name another file of the project shares: calls through that "
            "name may not be followed to the right function.",
        ),
        "unsupported-syntax": RuleDescription(
            "Unsupported syntax",
            "A function the analyzer could not analyze, because of syntax it does not support: its code is not checked.",
        ),
        "exploitable-vulnerability": RuleDescription(
            "Exploitable dependency vulnerability",
            "Attacker-controlled data reaches an API that an advisory of a vulnerable requirement affects.",
        ),
    }
)


def rule_descriptions(registry: PluginRegistry) -> Mapping[str, RuleDescription]:
    """What every rule of a check detects, for its report: the rules the engine reports
    itself, and those each loaded plugin describes, the first declaration of an id kept."""

    described: dict[str, RuleDescription] = dict(ENGINE_RULES)
    for loaded in registry:
        for rule_id, description in loaded.plugin.rules.items():
            described.setdefault(rule_id, description)
    return MappingProxyType(described)

ALL_ANALYSES: tuple[AnyAnalysis, ...] = (
    *SEMANTIC_ANALYSES,
    CFGAnalysis,
    DominanceAnalysis,
    PostDominanceAnalysis,
    PyIRAnalysis,
    ModuleIRAnalysis,
    SSAAnalysis,
    DefUseAnalysis,
    ConstantPropagation,
    RangeAnalysis,
    ModuleStringsAnalysis,
    ModuleCollectionsAnalysis,
    HeapAnalysis,
    CallGraphAnalysis,
    SummaryAnalysis,
    ProjectSummaries,
    SecurityModelAnalysis,
    TaintAnalysis,
    EntryPointAnalysis,
    RefutationAnalysis,
    DependencyAnalysis,
    RegisteredRoutes,
    EscapedTemplates,
    ClearingAnalysis,
    TemplateCallsAnalysis,
    SignaturesAnalysis,
)


# Named so the set stays right whichever analyses a build registers.
_PROJECT_DEPENDANT_NAMES = frozenset(
    {"interprocedural.project", "interprocedural.summaries", "taint.flows", "taint.entry_points", "findings.refutation"}
)
_PROJECT_DEPENDANTS: frozenset[AnyAnalysis] = frozenset(
    a for a in ALL_ANALYSES if a.name in _PROJECT_DEPENDANT_NAMES
)


class ProjectSummariesUpdated(TransformationPass):
    """Invalidates every result that depends on the project-wide summary index."""

    name: ClassVar[str] = "project.summaries-updated"
    preserves: ClassVar[frozenset[AnyAnalysis]] = frozenset(
        a for a in ALL_ANALYSES if a not in _PROJECT_DEPENDANTS
    )

    @classmethod
    def run(cls, ctx: AnalysisContext) -> None:
        pass


class ResultsEvicted(TransformationPass):
    """Drops a module's PyIR and every derived result once its summaries, call sites and
    findings are extracted (§30); the semantic tables and the engine inputs stay."""

    name: ClassVar[str] = "project.results-evicted"
    preserves: ClassVar[frozenset[AnyAnalysis]] = frozenset(
        {
            *SEMANTIC_ANALYSES,
            SecurityModelAnalysis,
            ProjectSummaries,
            DependencyAnalysis,
            RegisteredRoutes,
            EscapedTemplates,
            ClearingAnalysis,
            TemplateCallsAnalysis,
            SignaturesAnalysis,
        }
    )

    @classmethod
    def run(cls, ctx: AnalysisContext) -> None:
        pass


def _no_rules() -> Mapping[str, RuleDescription]:
    return MappingProxyType({})


def _no_keys() -> Mapping[str, str]:
    return MappingProxyType({})


@dataclass(frozen=True)
class ProjectAnalysis:
    graph: ModuleGraph
    index: SummaryIndex
    findings: tuple[Finding, ...]
    dependencies: DependencyGraph = field(default_factory=DependencyGraph)
    keys: Mapping[str, str] = field(default_factory=_no_keys)
    reused: tuple[str, ...] = ()
    advisories: tuple[Advisory, ...] = ()
    coverage: Coverage = field(default_factory=Coverage)
    # Findings silenced by an inline ``# coretrace: ignore`` comment.
    suppressed: tuple[Finding, ...] = ()
    # Findings about an advisory the policy accepts, still evidence of what is reached.
    accepted: tuple[Finding, ...] = ()
    # The plugins and advisory files the result was produced with.
    components: tuple[Component, ...] = ()
    # What each rule of the check detects, for its report.
    rules: Mapping[str, RuleDescription] = field(default_factory=_no_rules)


@dataclass(frozen=True)
class FileAnalysis:
    findings: tuple[Finding, ...]
    coverage: Coverage
    suppressed: tuple[Finding, ...] = ()
    # What each rule of the check detects, for its report.
    rules: Mapping[str, RuleDescription] = field(default_factory=_no_rules)


def _text_of(sources: SourceManager) -> Callable[[SourceId], str | None]:
    """The text of a source by identifier: loaded sources first, then the file on disk."""

    def text_of(source_id: SourceId) -> str | None:
        try:
            return sources.get(source_id).text
        except KeyError:
            pass
        path = Path(str(source_id))
        try:
            return decode_text(path.read_bytes()) if path.is_file() else None
        except (OSError, UnicodeDecodeError):
            return None

    return text_of


def build_manager(
    module: nodes.Module, models: SecurityModelRegistry | None = None
) -> AnalysisManager:
    """A manager with every engine analysis registered and the engine inputs provided."""

    manager = _register_all(module)
    table = (models or SecurityModelRegistry()).freeze()
    manager.provide(SecurityModelAnalysis, table)
    manager.provide(MembersAnalysis, table.members_by_class())
    manager.provide(SignaturesAnalysis, table.signatures())
    manager.provide(ClearingAnalysis, table.clearing())
    manager.provide(ProjectSummaries, SummaryIndex())
    manager.provide(RegisteredRoutes, _routes_of(manager))
    return manager


def _routes_of(manager: AnalysisManager) -> dict[SymbolId, EntryPoint]:
    """The handlers one module registers, given the models already provided."""

    return registered_routes(
        manager.module,
        manager.get(ScopeAnalysis),
        manager.get(SymbolAnalysis),
        manager.get(SecurityModelAnalysis),
    )


def load_plugins(plugin_roots: Sequence[Path], manager: AnalysisManager) -> PluginRegistry:
    """Every plugin below the roots, each manifest loaded once even when roots overlap."""

    registry = PluginRegistry()
    seen: set[Path] = set()
    for root in plugin_roots:
        for loaded in discover_plugins(root, manager):
            if loaded.directory.resolve() not in seen:
                seen.add(loaded.directory.resolve())
                registry.add(loaded)
    return registry


def plugin_models(plugins: Iterable[Plugin], root: Path | None = None) -> ModelTable:
    """The models of ``plugins``, with those they read from the project at ``root``."""

    models = SecurityModelRegistry()
    for plugin in plugins:
        models.register(*plugin.models, origin=plugin.name)
        if root is not None:
            models.register(*plugin.project_models(root), origin=plugin.name)
    return models.freeze()


def check(source: SourceFile, plugin_roots: Sequence[Path]) -> tuple[Finding, ...]:
    """Run every plugin found under ``plugin_roots`` against one file."""

    return analyze_file(source, plugin_roots).findings


def analyze_file(source: SourceFile, plugin_roots: Sequence[Path], root: Path | None = None) -> FileAnalysis:
    """One file's findings and coverage, with every plugin found under ``plugin_roots``.
    The context of its findings is judged on its path from ``root``, the root of its
    report: by default its own directory, as the command line reports a file checked
    alone. A source that is not a file is judged on its name as given."""

    manager = _register_all(build_hir(source))
    registry = load_plugins(plugin_roots, manager)
    table = plugin_models(loaded.plugin for loaded in registry)
    manager.provide(SecurityModelAnalysis, table)
    manager.provide(MembersAnalysis, table.members_by_class())
    manager.provide(SignaturesAnalysis, table.signatures())
    manager.provide(ClearingAnalysis, table.clearing())
    manager.provide(ProjectSummaries, SummaryIndex())
    manager.provide(RegisteredRoutes, _routes_of(manager))
    reference = None if source.path is None else root or source.path.resolve().parent
    findings, supported = _check_module(manager, tuple(loaded.plugin for loaded in registry), reference)
    path = path_in_project(source.source_id, reference)
    findings = tuple(_situated(f, path if f.span.source_id == source.source_id else None) for f in findings)
    functions = len(analyzable_functions(manager.module))
    coverage = Coverage((FileCoverage(str(source.source_id), "analysed", functions, len(supported)),))
    kept, suppressed = partition(findings, lambda sid: source.text if sid == source.source_id else None)
    return FileAnalysis(kept, coverage, suppressed, rule_descriptions(registry))


def resolve_dependencies(root: Path, sources: SourceManager) -> DependencyGraph:
    """The requirements declared or pinned by the dependency files at ``root`` (§26)."""

    graph = DependencyGraph()
    candidates = sorted(root.glob("requirements*.txt")) + [root / name for name in DEPENDENCY_FILES]
    for path in candidates:
        if not path.is_file():
            continue
        try:
            text = decode_text(path.read_bytes())
        except (OSError, UnicodeDecodeError) as error:
            graph = graph.merge(DependencyGraph(errors=(f"{path}: {error}",)))
            continue
        graph = graph.merge(parse_dependencies(sources.add_source(str(path), text)))
    return graph


# ``ProcessPoolExecutor`` refuses more workers on Windows, where it waits on at most 63
# handles.
_WINDOWS_POOL_LIMIT = 61


def process_pool_size(jobs: int, platform: str | None = None) -> int:
    """The processes a pool asked for ``jobs`` holds: no more than ``platform`` (this one
    by default) allows."""

    if (platform or sys.platform) == "win32":
        return min(jobs, _WINDOWS_POOL_LIMIT)
    return jobs


def analyze_project(
    root: Path,
    plugin_roots: Sequence[Path] = (),
    plugins: Sequence[Plugin] = (),
    cache: ProjectCache | None = None,
    jobs: int = 1,
    advisory_files: Sequence[Path] = (),
    policy_file: Path | None = None,
) -> ProjectAnalysis:
    """Analyse every Python file under ``root`` with a shared summary index (§21) and the
    dependency graph of its manifests (§26). ``plugins`` adds plugin instances to the
    ones discovered under ``plugin_roots``. With a ``cache``, modules whose key is
    unchanged since a previous run are served from it (§11). Modules are scheduled by
    strongly connected components of the module graph, imports first; with ``jobs``
    above one the components of a wave are analysed in that many processes (§29).
    ``advisory_files`` add to the ``advisories.json`` at ``root``; ``policy_file``
    replaces the ``coretrace-policy.toml`` there.

    Every module is known by the name ``name_modules`` gives its file: its import name,
    or its path from the root when another file has that import name. Such files carry
    an ``ambiguous-module`` note and are covered as ``ambiguous``."""

    if jobs < 1:
        raise ValueError("jobs must be at least 1")
    # Module sources are loaded by their resolved path; every other file the project is
    # read from (manifests, configuration, advisories, policy) must be named the same way,
    # or the reports and baselines would see it relative to the working directory (#206).
    root = root.resolve()
    advisory_files = [path.resolve() for path in advisory_files]
    policy_file = None if policy_file is None else policy_file.resolve()
    sources = SourceManager()
    findings: list[Finding] = []
    dependencies = resolve_dependencies(root, sources)
    for error in dependencies.errors:
        findings.append(_note("syntax-error", error, sources.add_source(error.split(":")[0], ""), 1))
    advisory_paths = _advisory_paths(root, advisory_files)
    file_advisories: list[tuple[Path, tuple[Advisory, ...]]] = []
    for path in advisory_paths:
        try:
            file_advisories.append((path, load_advisories(path)))
        except AdvisoryFileError as error:
            findings.append(_note("syntax-error", str(error), sources.add_source(str(path), ""), 1))
    policy = Policy()
    policy_path = policy_file if policy_file is not None else root / POLICY_FILE
    if policy_path.is_file():
        try:
            policy = load_policy(policy_path)
        except AdvisoryFileError as error:
            findings.append(_note("syntax-error", str(error), sources.add_source(str(policy_path), ""), 1))
    modules: dict[str, nodes.Module] = {}
    files: dict[str, SourceFile] = {}
    unreadable: list[tuple[Path, str]] = []
    coverage: list[FileCoverage] = []
    named = name_modules(root, discover_sources(root, sources, unreadable))
    for path, reason in unreadable:
        findings.append(_note("syntax-error", f"{path}: {reason}", sources.add_source(str(path), ""), 1))
        coverage.append(FileCoverage(str(path), "unreadable", 0, 0))
    for name, source in named.items():
        try:
            modules[name] = build_hir(source)
        except (ParseError, HIRBuildError) as error:
            findings.append(_note("syntax-error", str(error), source, 1))
            coverage.append(FileCoverage(str(source.source_id), "syntax-error", 0, 0))
            continue
        files[name] = source

    managers = {name: _register_all(module) for name, module in modules.items()}
    probe = next(iter(managers.values()), None) or _register_all(build_hir(sources.add_source("<empty>", "")))
    registry = load_plugins(plugin_roots, probe)
    all_plugins: tuple[Plugin, ...] = (*(loaded.plugin for loaded in registry), *plugins)
    components = _components(registry, root, (path for path, _ in file_advisories))
    advisories, origins = _merge_advisories(_contributions(registry, plugins, root, file_advisories))
    affected = affected_symbols(dependencies, advisories)
    models = plugin_models(all_plugins, root).extended(
        *advisory_sinks(affected), suffixes=advisory_suffix_sinks(affected)
    )
    members = models.members_by_class()
    for manager in managers.values():
        manager.provide(SecurityModelAnalysis, models)
        manager.provide(MembersAnalysis, members)
        manager.provide(SignaturesAnalysis, models.signatures())
        manager.provide(DependencyAnalysis, dependencies)

    imports: dict[str, ImportTable] = {}
    analysable: dict[str, AnalysisManager] = {}
    for name, manager in managers.items():
        try:
            imports[name] = manager.get(ImportAnalysis)
        except (ImportResolutionError, ScopeError) as error:
            findings.append(_note("syntax-error", str(error), files[name], 1))
            coverage.append(FileCoverage(str(files[name].source_id), "syntax-error", 0, 0))
            continue
        analysable[name] = manager
    graph = build_module_graph(
        {name: files[name] for name in analysable}, {name: modules[name] for name in analysable}, imports
    )
    shared = _shared_names(graph)
    findings.extend(_ambiguity_notes(graph, shared, root))
    routes: dict[SymbolId, EntryPoint] = {}
    for name in sorted(analysable):
        for symbol, registered in _routes_of(analysable[name]).items():
            routes.setdefault(symbol, registered)
    escaped = escaped_templates(root)
    templates = project_templates(root)
    clearing = models.clearing(escaped)
    template_calls = models.template_calls(
        templates.filters,
        request_processor((modules[name], analysable[name].get(ScopeAnalysis)) for name in sorted(analysable)),
    )
    for manager in analysable.values():
        manager.provide(RegisteredRoutes, routes)
        manager.provide(EscapedTemplates, escaped)
        manager.provide(ClearingAnalysis, clearing)
        manager.provide(TemplateCallsAnalysis, template_calls)

    configuration = _configuration_key(
        components, plugins, models, advisories, dependencies, routes, escaped, template_calls
    )
    # A module's names are part of its key: under another name, what the modules importing
    # it reach, and the symbols of its own functions, change.
    keys = module_keys(
        graph,
        {
            name: fingerprint(configuration, str(files[name].source_id), name, *_names_of(files[name]), files[name].text)
            for name in analysable
        },
    )
    results: dict[str, CachedModule] = {}
    if cache is not None:
        for name in analysable:
            entry = cache.load(keys[name])
            if entry is not None:
                results[name] = entry
    reused = tuple(sorted(results))

    module_plugins = tuple(p for p in all_plugins if not isinstance(p, ProjectPlugin))
    suffixes = suffix_index(affected, analysable)
    pool = None
    spooled: tempfile.TemporaryDirectory[str] | None = None
    # Where finished modules are kept for the next waves: the cache, else a directory of
    # this run's own. Workers read from it what a component imports (#245).
    spool = cache
    if jobs > 1:
        if spool is None:
            spooled = tempfile.TemporaryDirectory(prefix="coretrace-")
            spool = ProjectCache(Path(spooled.name))
        project = _Project(
            root,
            tuple(plugin_roots),
            tuple(plugins),
            advisory_paths,
            _encode_routes(routes),
            tuple(sorted(escaped)),
            template_calls,
            tuple(sorted(analysable)),
            spool.directory,
        )
        pool = concurrent.futures.ProcessPoolExecutor(
            max_workers=process_pool_size(jobs),
            mp_context=multiprocessing.get_context("spawn"),
            initializer=_start_worker,
            initargs=(project,),
        )
    try:
        for wave in graph.schedule():
            pending = [c for c in wave if any(m not in results for m in c)]
            computed: dict[str, CachedModule] = {}
            if pool is None:
                for component in pending:
                    batch = {name: analysable[name] for name in sorted(component)}
                    computed.update(
                        _analyse_managers(
                            batch,
                            results,
                            _imported_names(graph, component, results),
                            module_plugins,
                            affected,
                            suffixes,
                            root,
                        )
                    )
                    for manager in batch.values():
                        manager.run(ResultsEvicted)
            else:
                futures = []
                for component in pending:
                    names = _imported_names(graph, component, results)
                    futures.append(
                        pool.submit(
                            _analyse_batch,
                            _Batch(
                                {name: _path_of(files[name]) for name in sorted(component)},
                                {name: keys[name] for read in names.values() for name in read if name not in component},
                                names,
                            ),
                        )
                    )
                for future in futures:
                    computed.update({name: decode(data) for name, data in future.result().items()})
            store = cache if pool is None else spool
            for name, entry in computed.items():
                if store is not None:
                    store.store(keys[name], entry)
            results.update(computed)
    finally:
        if pool is not None:
            pool.shutdown()
        if spooled is not None:
            spooled.cleanup()

    index = _index(results, {name: (name,) for name in results})
    call_graphs: dict[str, CallGraph] = {}
    functions: dict[str, tuple[ModuleFunction, ...]] = {}
    for name in sorted(analysable):
        entry = results[name]
        findings.extend(entry.findings)
        unsupported = sum(1 for f in entry.findings if f.rule_id == "unsupported-syntax")
        status = "ambiguous" if files[name].module_name in shared else "analysed"
        coverage.append(
            FileCoverage(str(files[name].source_id), status, len(entry.functions), len(entry.functions) - unsupported)
        )
        functions[name] = entry.functions
        sites: dict[str, list[CallSite]] = {function.name: [] for function in entry.functions}
        for site in entry.sites:
            sites.setdefault(site.caller, []).append(site)
        reads: dict[str, list[SymbolRead]] = {}
        for read in entry.reads:
            reads.setdefault(read.function, []).append(read)
        call_graphs[name] = CallGraph(
            {},
            {f: tuple(s) for f, s in sites.items()},
            frozenset(),
            reads={f: tuple(r) for f, r in reads.items()},
        )
    context = ProjectContext(
        graph, dependencies, advisories, analysable, call_graphs, policy, root, functions, templates
    )
    for plugin in all_plugins:
        if isinstance(plugin, ProjectPlugin):
            findings.extend(plugin.analyze_project(context))
    # The context comes first, so that a refiner can read it.
    findings = [_situated(finding, path_in_project(finding.span.source_id, root)) for finding in findings]
    for plugin in all_plugins:
        if isinstance(plugin, ProjectPlugin):
            findings = list(apply_refinement(plugin, findings, plugin.refine(context, tuple(findings))))
    findings = [_sourced(finding, origins) for finding in findings]
    accepted = tuple(f for f in findings if policy.accepts(f))
    kept, suppressed = partition(apply_policy(policy, findings), _text_of(sources))
    unread = list(templates.unread)
    for name in sorted(call_graphs):
        unread.extend(unread_renders(name, call_graphs[name], models, templates.names))
    return ProjectAnalysis(
        graph,
        index,
        kept,
        dependencies,
        MappingProxyType(keys),
        reused,
        advisories,
        Coverage(tuple(sorted(coverage, key=lambda c: c.path)), tuple(unread)),
        suppressed,
        accepted,
        components,
        rule_descriptions(registry),
    )


Origins = Mapping[tuple[str, str], str]


def _merge_advisories(contributions: Iterable[tuple[str, Iterable[Advisory]]]) -> tuple[tuple[Advisory, ...], Origins]:
    """One advisory per identifier and package, with the contributor it comes from: a
    later contributor replaces an earlier one, so a curated plugin refines a bundled
    sample and a local file, the project's own feed, wins over every plugin."""

    merged: dict[tuple[str, str], tuple[str, Advisory]] = {}
    for source, advisories in contributions:
        for advisory in advisories:
            merged[(advisory.id, advisory.package)] = (source, advisory)
    return (
        tuple(advisory for _, advisory in merged.values()),
        MappingProxyType({key: source for key, (source, _) in merged.items()}),
    )


def _contributions(
    registry: PluginRegistry,
    plugins: Sequence[Plugin],
    root: Path,
    files: Iterable[tuple[Path, tuple[Advisory, ...]]],
) -> list[tuple[str, Iterable[Advisory]]]:
    """The advisories of each plugin, then of each advisory file, each under the name a
    finding cites it by: ``name@version`` for a loaded plugin, the path for a file."""

    return [
        *((f"{loaded.manifest.name}@{loaded.manifest.version}", loaded.plugin.advisories) for loaded in registry),
        *((plugin.name, plugin.advisories) for plugin in plugins),
        *((_located(path, root), advisories) for path, advisories in files),
    ]


def _components(registry: PluginRegistry, root: Path, files: Iterable[Path]) -> tuple[Component, ...]:
    """The plugins and advisory files a project result is produced with, each with the
    digest of its content: the plugin directory as the cache fingerprints it, the file's
    bytes."""

    return (
        *(
            Component(PLUGIN, loaded.manifest.name, f"sha256:{directory_fingerprint(loaded.directory)}", loaded.manifest.version)
            for loaded in registry
        ),
        *(
            Component(ADVISORIES, _located(path, root), f"sha256:{hashlib.sha256(path.read_bytes()).hexdigest()}")
            for path in files
        ),
    )


def _located(path: Path, root: Path) -> str:
    """``path`` relative to ``root``, POSIX style, when it lies under it."""

    try:
        return path.resolve().relative_to(root.resolve()).as_posix()
    except ValueError:
        return str(path)


def _situated(finding: Finding, path: str | None) -> Finding:
    """``finding`` with the context (``code_context``) and the origin (``code_origin``) of
    its code in its metadata, judged on ``path``, its file's path from the root of the
    report. A context the rule gave the finding stays: it is the rule's (#244). The origin
    is told apart from the context, so vendored test code is both (#243)."""

    if path is None:
        return finding
    added: dict[str, str] = {}
    context = code_context(path)
    if context is not None and "context" not in finding.metadata:
        added["context"] = context
    origin = code_origin(path)
    if origin is not None:
        added["origin"] = origin
    return replace(finding, metadata={**finding.metadata, **added}) if added else finding


def _sourced(finding: Finding, origins: Origins) -> Finding:
    """``finding`` with the plugin or advisory file its advisory comes from, when it
    cites one."""

    origin = origins.get((finding.metadata.get("advisory", ""), finding.metadata.get("package", "")))
    if origin is None:
        return finding
    return replace(finding, metadata={**finding.metadata, "advisory_source": origin})


def _advisory_paths(root: Path, advisory_files: Sequence[Path]) -> tuple[Path, ...]:
    """The advisory file at ``root``, when present, then the explicit ones."""

    default = root / ADVISORY_FILE
    return (*([default] if default.is_file() else []), *advisory_files)


def _shared_names(graph: ModuleGraph) -> dict[str, list[str]]:
    """The module names several analysable files have, each with those files' modules.
    A name a module also has from its own directory (``import_names``) is not among
    them: the importer's root tells such files apart."""

    by_name: dict[str, list[str]] = {}
    for name in graph.modules:
        by_name.setdefault(graph.source(name).module_name, []).append(name)
    return {import_name: names for import_name, names in by_name.items() if len(names) > 1}


def _ambiguity_notes(graph: ModuleGraph, shared: Mapping[str, list[str]], root: Path) -> list[Finding]:
    """One note on each file whose import name another file shares, since a symbol of
    that name may denote a function of either, and one on each importer whose import of
    such a name no root resolved, since calls through it are not followed."""

    def located(name: str) -> str:
        return _located(_path_of(graph.source(name)), root)

    having: dict[str, list[str]] = {}
    for module in graph.modules:
        for import_name, _ in import_names(graph.source(module)):
            having.setdefault(import_name, []).append(module)
    notes: list[Finding] = []
    for import_name, names in sorted(shared.items()):
        for name in names:
            others = ", ".join(located(other) for other in names if other != name)
            message = (
                f"{located(name)} is the module '{import_name}' like {others}: a symbol "
                f"python.{import_name}.* may name a function of either, and an import of "
                f"'{import_name}' resolves only from its own root"
            )
            notes.append(_note("ambiguous-module", message, graph.source(name), 1))
    for name in graph.modules:
        for import_name in sorted(graph.unresolved(name)):
            candidates = " or ".join(located(other) for other in having[import_name])
            message = f"'{import_name}' imported here is {candidates}: calls into it are not followed"
            notes.append(_note("ambiguous-module", message, graph.source(name), 1))
    return notes


def _imported_names(
    graph: ModuleGraph, component: frozenset[str], results: Mapping[str, CachedModule]
) -> dict[str, dict[str, tuple[str, ...]]]:
    """For each member of a component, the modules whose summaries it reads, each with
    the import names the member's own code refers to it by: the members, and every
    module they import, transitively, all final by the time the component's wave runs.
    The member goes under its own names, each of which names it from its own roots.
    Another module goes under the names the member resolved it by, and otherwise under
    its module name only, the name the whole project knows it by (``pkg.helpers``,
    reached through ``import pkg``), never under a name another module resolved from
    its own directory: what a callee, or another member of a cycle, resolved does not
    change what the member's names mean. Of several such modules with one module name,
    only one the member imports directly goes under it, and a name the member left
    unresolved goes under none."""

    reached = set(component)
    pending = list(component)
    while pending:
        for imported in graph.imports(pending.pop()):
            if imported in results and imported not in reached:
                reached.add(imported)
                pending.append(imported)
    by_module_name: dict[str, list[str]] = {}
    for name in sorted(reached):
        by_module_name.setdefault(graph.source(name).module_name, []).append(name)
    read: dict[str, dict[str, tuple[str, ...]]] = {}
    for member in sorted(component):
        # The graph records no import of a module by itself.
        chosen = {
            **dict.fromkeys(_names_of(graph.source(member)), member),
            **{import_name: module for import_name, module in graph.resolved(member).items() if module in reached},
        }
        direct = graph.imports(member)
        for module_name, modules in by_module_name.items():
            picked = modules if len(modules) == 1 else [name for name in modules if name in direct]
            if module_name not in chosen and module_name not in graph.unresolved(member) and len(picked) == 1:
                chosen[module_name] = picked[0]
        names: dict[str, list[str]] = {}
        for import_name, module in sorted(chosen.items()):
            names.setdefault(module, []).append(import_name)
        read[member] = {module: tuple(module_names) for module, module_names in names.items()}
    return read


def _index(results: Mapping[str, CachedModule], names: Mapping[str, Iterable[str]]) -> SummaryIndex:
    """The summaries of the modules in ``names``, each function under the project symbol
    of every module name given for its module: one module, reached by several names,
    has its summaries computed once."""

    return SummaryIndex(
        {
            project_symbol(module, function): summary
            for name, module_names in sorted(names.items())
            for module in module_names
            for function, summary in results[name].summaries.items()
        }
    )


def _names_of(source: SourceFile) -> tuple[str, ...]:
    """The import names of ``source`` (``import_names``), without their roots."""

    return tuple(name for name, _ in import_names(source))


def _analyse_managers(
    managers: Mapping[str, AnalysisManager],
    imported: Mapping[str, CachedModule],
    names: Mapping[str, Mapping[str, tuple[str, ...]]],
    plugins: tuple[Plugin, ...],
    affected: Affected,
    suffixes: SuffixIndex,
    root: Path,
) -> dict[str, CachedModule]:
    """Analyse one component: iterate its summaries to a fixpoint (§21), then extract
    what the rest of the run needs from each module. Each member reads, in an index of
    its own, the modules ``names`` gives it under the names given (``_imported_names``):
    the members' summaries as they grow, the others' as ``imported`` holds them, final.
    Members that read every module under the same names, as in a regular package, share
    one index."""

    views = {member: tuple(sorted(names[member].items())) for member in managers}
    final = {
        view: _index(imported, {name: module_names for name, module_names in view if name not in managers})
        for view in dict.fromkeys(views.values())
    }

    def indexes(
        summaries: Mapping[str, Mapping[str, FunctionSummary]],
    ) -> dict[tuple[tuple[str, tuple[str, ...]], ...], SummaryIndex]:
        return {
            view: SummaryIndex(
                {
                    **seed.summaries,
                    **{
                        project_symbol(module, function): summary
                        for name, module_names in view
                        if name in managers
                        for module in module_names
                        for function, summary in summaries[name].items()
                    },
                }
            )
            for view, seed in final.items()
        }

    # The members' functions start from the summaries ``SummaryAnalysis`` starts from.
    # Left out, a call from one member to another is a call to an unknown symbol on the
    # first iteration and a project call on the next: around a cycle of modules, the
    # iteration then shifts by one module each time and never settles (#261).
    index = indexes({name: starting_summaries(manager.get(CallGraphAnalysis)) for name, manager in managers.items()})
    for name, manager in managers.items():
        manager.provide(ProjectSummaries, index[views[name]])
    # Iterated until every member's index settles, with no bound: stopped earlier,
    # summaries that started empty would miss what later iterations add, a dependency
    # still on its way reading as no taint. It settles because each iteration can only
    # add: each index holds the same symbols from the start, and in every round of a
    # module's fixpoint the rules that read the index join what they read (a call's
    # dependencies and records, heap stores, merges), while those that do not join (a
    # nonlocal record not summarized yet, ``kept``, reachability, ``unsupported``,
    # ``static``) depend on the code and the round only. Each round's table thus grows
    # with the index, the fixpoint too, in a finite domain: the parameters, symbols and
    # call records of the code. A rule reading the index without joining would break
    # this, and the loop's termination with it.
    while True:
        updated = indexes({name: _summaries_of(manager) for name, manager in managers.items()})
        if updated == index:
            break
        index = updated
        for name, manager in managers.items():
            manager.run(ProjectSummariesUpdated)
            manager.provide(ProjectSummaries, index[views[name]])
    return {name: _analyse_module(manager, plugins, affected, suffixes, root) for name, manager in managers.items()}


@dataclass(frozen=True)
class _Project:
    """What every worker process needs of the project, whatever the component it
    analyses: it rebuilds the configuration from the project root and the plugin roots
    once, when the pool starts it (``_start_worker``), and reads the finished modules
    from ``spool``, a ``ProjectCache`` directory the parent writes after each wave."""

    root: Path
    plugin_roots: tuple[Path, ...]
    plugins: tuple[Plugin, ...]
    advisory_paths: tuple[Path, ...]
    routes: tuple[tuple[str, str, str, int], ...]
    escaped: tuple[str, ...]
    template_calls: TemplateCalls
    # Every analysable module of the project, not only a component's: a curated
    # entry-point suffix matches derived symbols of any project module.
    modules: tuple[str, ...]
    spool: Path


@dataclass(frozen=True)
class _Batch:
    """One component handed to a worker process: its files under the engine's module
    names, the key in the spool of each module it imports, and the names each member
    reads each module by (``_imported_names``). Only these travel, not the summaries
    themselves."""

    paths: Mapping[str, Path]
    keys: Mapping[str, str]
    names: Mapping[str, Mapping[str, tuple[str, ...]]]


@dataclass
class _Worker:
    """The configuration a worker process prepared once, and the finished modules it has
    read from the spool so far, by key."""

    managed: Callable[[AnalysisManager], None]
    plugins: tuple[Plugin, ...]
    affected: Affected
    suffixes: SuffixIndex
    spool: ProjectCache
    root: Path
    read: dict[str, CachedModule] = field(default_factory=dict)

    def entries(self, keys: Mapping[str, str]) -> dict[str, CachedModule]:
        found: dict[str, CachedModule] = {}
        for name, key in keys.items():
            entry = self.read.get(key)
            if entry is None:
                entry = self.spool.load(key)
                if entry is None:
                    raise RuntimeError(f"the summaries of {name} are missing from {self.spool.directory}")
                self.read[key] = entry
            found[name] = entry
        return found


# The state of a worker process, set by ``_start_worker`` when the pool starts it.
_WORKER: _Worker | None = None


def _start_worker(project: _Project) -> None:
    """Prepare this worker process for every component it will analyse (#245)."""

    global _WORKER
    sources = SourceManager()
    probe = _register_all(build_hir(sources.add_source("<empty>", "")))
    registry = load_plugins(project.plugin_roots, probe)
    all_plugins: tuple[Plugin, ...] = (*(loaded.plugin for loaded in registry), *project.plugins)
    dependencies = resolve_dependencies(project.root, sources)
    file_advisories: list[tuple[Path, tuple[Advisory, ...]]] = []
    for path in project.advisory_paths:
        try:
            file_advisories.append((path, load_advisories(path)))
        except AdvisoryFileError:
            continue
    advisories, _ = _merge_advisories(_contributions(registry, project.plugins, project.root, file_advisories))
    affected = affected_symbols(dependencies, advisories)
    models = plugin_models(all_plugins, project.root).extended(
        *advisory_sinks(affected), suffixes=advisory_suffix_sinks(affected)
    )
    routes = _decode_routes(project.routes)
    escaped = frozenset(project.escaped)
    members = models.members_by_class()
    signatures = models.signatures()
    clearing = models.clearing(escaped)

    def managed(manager: AnalysisManager) -> None:
        manager.provide(SecurityModelAnalysis, models)
        manager.provide(MembersAnalysis, members)
        manager.provide(SignaturesAnalysis, signatures)
        manager.provide(DependencyAnalysis, dependencies)
        manager.provide(RegisteredRoutes, routes)
        manager.provide(EscapedTemplates, escaped)
        manager.provide(ClearingAnalysis, clearing)
        manager.provide(TemplateCallsAnalysis, project.template_calls)

    _WORKER = _Worker(
        managed,
        tuple(p for p in all_plugins if not isinstance(p, ProjectPlugin)),
        affected,
        suffix_index(affected, project.modules),
        ProjectCache(project.spool),
        project.root,
    )


def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]:
    worker = _WORKER
    assert worker is not None, "the pool starts every worker with _start_worker"
    sources = {name: SourceManager().load_file(path, worker.root) for name, path in batch.paths.items()}
    managers = {name: _register_all(build_hir(source)) for name, source in sources.items()}
    for manager in managers.values():
        worker.managed(manager)
    results = _analyse_managers(
        managers,
        worker.entries(batch.keys),
        batch.names,
        worker.plugins,
        worker.affected,
        worker.suffixes,
        worker.root,
    )
    return {name: encode(entry) for name, entry in results.items()}


def _path_of(source: SourceFile) -> Path:
    assert source.path is not None, "project sources are loaded from files"
    return source.path


def _encode_routes(routes: Routes) -> tuple[tuple[str, str, str, int], ...]:
    return tuple(
        sorted((str(symbol), str(entry.symbol), entry.label, entry.kinds.value) for symbol, entry in routes.items())
    )


def _decode_routes(data: tuple[tuple[str, str, str, int], ...]) -> dict[SymbolId, EntryPoint]:
    return {SymbolId(symbol): EntryPoint(SymbolId(registrar), label, TaintKind(kinds)) for symbol, registrar, label, kinds in data}


def _configuration_key(
    components: Sequence[Component],
    plugins: Sequence[Plugin],
    models: ModelTable,
    advisories: tuple[Advisory, ...],
    dependencies: DependencyGraph,
    routes: Routes | None = None,
    escaped: frozenset[str] = frozenset(),
    template_calls: TemplateCalls | None = None,
) -> str:
    """Everything a module's results depend on besides the project sources (§11)."""

    return fingerprint(
        __version__,
        str(HIR_SCHEMA_VERSION),
        str(PYIR_SCHEMA_VERSION),
        str(FINDING_SCHEMA_VERSION),
        str(PLUGIN_API_VERSION),
        *(f"{c.name}={c.version}:{c.digest}" for c in components if c.kind == PLUGIN),
        *(f"{type(p).__module__}.{type(p).__qualname__}" for p in plugins),
        repr(models),
        repr(advisories),
        repr(dependencies.requirements),
        repr(dependencies.errors),
        repr(_encode_routes(routes or {})),
        repr(sorted(escaped)),
        repr(template_calls),
    )


def _analyse_module(
    manager: AnalysisManager, plugins: tuple[Plugin, ...], affected: Affected, suffixes: SuffixIndex, root: Path
) -> CachedModule:
    """One module's findings, summaries, call sites and symbol reads: what the cache
    keeps (§11)."""

    findings, supported = _check_module(manager, plugins, root)
    graph = manager.get(CallGraphAnalysis)
    correlated: list[Finding] = []
    if affected:
        strings = manager.get(ModuleStringsAnalysis)
        for function in supported:
            flows = manager.get(TaintAnalysis, function).flows
            ssa = manager.get(SSAAnalysis, function)
            defs = {i.result: i for block in ssa.blocks for i in block.instructions if i.result is not None}
            urls = {flow: flow_url(flow, defs, strings) for flow in flows if flow.kinds & TaintKind.ADVISORY}
            name = graph.name_of(function)
            # A flow through a callee reaches the sink at another call: no receiver here.
            prior = {
                flow: graph.prior_calls_at(name, flow.location)
                for flow in flows
                if flow.kinds & TaintKind.ADVISORY and flow.through is None
            }
            correlated.extend(
                correlate(
                    name, flows, manager.get(RefutationAnalysis, function), affected, urls, suffixes, prior
                )
            )
    return CachedModule(
        manager.get(EntryPointAnalysis),
        _summaries_of(manager),
        tuple(site for function in graph.functions for site in graph.sites(function)),
        (*findings, *correlated),
        tuple(read for function in graph.functions for read in graph.reads(function)),
    )


def _summaries_of(manager: AnalysisManager) -> dict[str, FunctionSummary]:
    table = manager.get(SummaryAnalysis)
    return {name: table.summary(name) for name in table.names}


def _check_module(
    manager: AnalysisManager, plugins: tuple[Plugin, ...], root: Path | None
) -> tuple[tuple[Finding, ...], tuple[nodes.Function, ...]]:
    """Findings of the module plugins plus notes for unsupported functions, and the
    functions that could be analysed; ``root`` is the root of the analysis the plugins
    read a file's path in the project from (``PluginContext.path``)."""

    supported: list[nodes.Function] = []
    notes: list[Finding] = []
    for function in analyzable_functions(manager.module):
        try:
            manager.get(SSAAnalysis, function)
        except (LoweringError, CFGError) as error:
            notes.append(
                Finding(
                    rule_id="unsupported-syntax",
                    message=str(error),
                    severity=Severity.INFO,
                    confidence=Confidence.HIGH,
                    span=function.span,
                    function=function.name,
                )
            )
        else:
            supported.append(function)
    findings = run_plugins(manager, plugins, tuple(supported), root)
    return (*findings, *notes), tuple(supported)


def _note(rule_id: str, message: str, source: SourceFile, line: int) -> Finding:
    return Finding(
        rule_id=rule_id,
        message=message,
        severity=Severity.INFO,
        confidence=Confidence.HIGH,
        span=SourceSpan(source.source_id, line, 1),
    )


def _register_all(module: nodes.Module) -> AnalysisManager:
    manager = AnalysisManager(module)
    manager.register(*ALL_ANALYSES)
    return manager


def report(
    findings: Sequence[Finding],
    coverage: Coverage | None = None,
    root: Path | None = None,
    suppressed: Sequence[Finding] = (),
    baselined: Sequence[Finding] | None = None,
    components: Sequence[Component] = (),
    rules: Mapping[str, RuleDescription] | None = None,
) -> Report:
    """The report of a check; paths under ``root`` are rendered relative to it.
    ``baselined`` is the findings a baseline accounts for, ``None`` when none applied."""

    return Report(
        tuple(findings),
        TOOL_NAME,
        __version__,
        coverage,
        root,
        tuple(suppressed),
        tuple(baselined or ()),
        baselined is not None,
        tuple(components),
        MappingProxyType(dict(rules or {})),
    )