19381742489

Committed 15 Nov 2025 12:52AM UTC coverage: 49.706% (-30.6%) from 80.29%

Build # 19381742489

Build Type

Pull #22890

github

Committed by

web-flow

Commit Message

Merge d961abf79 into 42e1ebd41

Pull Request Pull Request #22890: Updated all python subsystem constraints to 3.14

Run Details

4 of 5 new or added lines in 5 files covered. (80.0%)

14659 existing lines in 485 files now uncovered.

31583 of 63540 relevant lines covered (49.71%)

0.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

58.7

/src/python/pants/backend/tools/semgrep/rules.py

# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations

import itertools
import logging
from collections import defaultdict
from collections.abc import Iterable
from dataclasses import dataclass
from pathlib import PurePath

from pants.backend.python.util_rules import pex
from pants.backend.python.util_rules.pex import VenvPexProcess, create_venv_pex
from pants.core.goals.lint import LintResult, LintTargetsRequest
from pants.core.util_rules.partitions import Partition, Partitions
from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files
from pants.engine.addresses import Address
from pants.engine.fs import CreateDigest, FileContent, MergeDigests, PathGlobs, Paths
from pants.engine.intrinsics import (
    create_digest,
    digest_to_snapshot,
    execute_process,
    merge_digests,
    path_globs_to_paths,
)
from pants.engine.process import ProcessCacheScope
from pants.engine.rules import Rule, collect_rules, concurrently, implicitly, rule
from pants.engine.unions import UnionRule
from pants.option.global_options import GlobalOptions
from pants.util.logging import LogLevel
from pants.util.strutil import pluralize

from .subsystem import SemgrepFieldSet, SemgrepSubsystem

logger = logging.getLogger(__name__)


_SEMGREPIGNORE_FILE_NAME = ".semgrepignore"
_DEFAULT_SEMGREP_CONFIG_DIR = ".semgrep"


class SemgrepLintRequest(LintTargetsRequest):
    field_set_type = SemgrepFieldSet
    tool_subsystem = SemgrepSubsystem  # type: ignore[assignment]


@dataclass(frozen=True)
class PartitionMetadata:
    config_files: frozenset[PurePath]

    @property
    def description(self) -> str:
        return ", ".join(sorted(str(path) for path in self.config_files))


@dataclass
class AllSemgrepConfigs:
    configs_by_dir: dict[PurePath, set[PurePath]]

    def ancestor_configs(self, address: Address) -> Iterable[PurePath]:
        # TODO: introspect the semgrep rules and determine which (if any) apply to the files, e.g. a
        # Python file shouldn't depend on a .semgrep.yml that doesn't have any 'python' or 'generic'
        # rules, and similarly if there's path inclusions/exclusions.
        # TODO: this would be better as actual dependency inference (e.g. allows inspection, manual
        # addition/exclusion), but that can only infer 'full' dependencies and it is wrong (e.g. JVM
        # things break) for real code files to depend on this sort of non-code linter config; requires
        # dependency scopes or similar (https://github.com/pantsbuild/pants/issues/12794)
        spec = PurePath(address.spec_path)

        for ancestor in itertools.chain([spec], spec.parents):
            yield from self.configs_by_dir.get(ancestor, [])


def _group_by_semgrep_dir(
    all_config_files: Paths, all_config_dir_files: Paths, config_name: str
) -> AllSemgrepConfigs:
    configs_by_dir: dict[PurePath, set[PurePath]] = {}
    for config_path in all_config_files.files:
        # Rules like foo/semgrep.yaml should apply to the project at foo/
        path = PurePath(config_path)
        configs_by_dir.setdefault(path.parent, set()).add(path)

    for config_path in all_config_dir_files.files:
        # Rules like foo/bar/.semgrep/baz.yaml and foo/bar/.semgrep/baz/qux.yaml should apply to the
        # project at foo/bar/
        path = PurePath(config_path)
        config_directory = next(
            parent.parent for parent in path.parents if parent.name == config_name
        )
        configs_by_dir.setdefault(config_directory, set()).add(path)

    return AllSemgrepConfigs(configs_by_dir)


@rule
async def find_all_semgrep_configs(semgrep: SemgrepSubsystem) -> AllSemgrepConfigs:
    config_file_globs: tuple[str, ...] = ()
    config_dir_globs: tuple[str, ...] = ()

    if semgrep.config_name is None:
        config_file_globs = ("**/.semgrep.yml", "**/.semgrep.yaml")
        config_dir_globs = (
            f"**/{_DEFAULT_SEMGREP_CONFIG_DIR}/**/*.yaml",
            f"**/{_DEFAULT_SEMGREP_CONFIG_DIR}/**/*.yml",
        )
    elif semgrep.config_name.endswith((".yaml", ".yml")):
        config_file_globs = (f"**/{semgrep.config_name}",)
    else:
        config_dir_globs = (
            f"**/{semgrep.config_name}/**/*.yaml",
            f"**/{semgrep.config_name}/**/*.yml",
        )

    all_config_files = await path_globs_to_paths(PathGlobs(config_file_globs))
    all_config_dir_files = await path_globs_to_paths(PathGlobs(config_dir_globs))
    return _group_by_semgrep_dir(
        all_config_files,
        all_config_dir_files,
        (semgrep.config_name or _DEFAULT_SEMGREP_CONFIG_DIR),
    )


@dataclass(frozen=True)
class RelevantSemgrepConfigsRequest:
    field_set: SemgrepFieldSet


class RelevantSemgrepConfigs(frozenset[PurePath]):
    pass


@rule
async def infer_relevant_semgrep_configs(
    request: RelevantSemgrepConfigsRequest, all_semgrep: AllSemgrepConfigs
) -> RelevantSemgrepConfigs:
    return RelevantSemgrepConfigs(all_semgrep.ancestor_configs(request.field_set.address))


@rule
async def partition(
    request: SemgrepLintRequest.PartitionRequest[SemgrepFieldSet],
    semgrep: SemgrepSubsystem,
) -> Partitions:
    if semgrep.skip:
        return Partitions()

    all_configs = await concurrently(
        infer_relevant_semgrep_configs(RelevantSemgrepConfigsRequest(field_set), **implicitly())
        for field_set in request.field_sets
    )

    # partition by the sets of configs that apply to each input
    by_config = defaultdict(list)
    for field_set, configs in zip(request.field_sets, all_configs):
        if configs:
            by_config[configs].append(field_set)

    return Partitions(
        Partition(tuple(field_sets), PartitionMetadata(configs))
        for configs, field_sets in by_config.items()
    )


# We have a hard-coded settings file to side-step
# https://github.com/returntocorp/semgrep/issues/7102, and also provide more cacheability, NB. both
# keys are required.
_DEFAULT_SETTINGS = FileContent(
    path="__semgrep_settings.yaml",
    content=b"anonymous_user_id: 00000000-0000-0000-0000-000000000000\nhas_shown_metrics_notification: true",
)


@rule(desc="Lint with Semgrep", level=LogLevel.DEBUG)
async def lint(
    request: SemgrepLintRequest.Batch[SemgrepFieldSet, PartitionMetadata],
    semgrep: SemgrepSubsystem,
    global_options: GlobalOptions,
) -> LintResult:
    config_files, ignore_files, semgrep_pex, input_files, settings = await concurrently(
        digest_to_snapshot(
            **implicitly(PathGlobs(str(s) for s in request.partition_metadata.config_files))
        ),
        digest_to_snapshot(**implicitly(PathGlobs([_SEMGREPIGNORE_FILE_NAME]))),
        create_venv_pex(**implicitly(semgrep.to_pex_request())),
        determine_source_files(
            SourceFilesRequest(field_set.source for field_set in request.elements)
        ),
        create_digest(CreateDigest([_DEFAULT_SETTINGS])),
    )

    input_digest = await merge_digests(
        MergeDigests(
            (
                input_files.snapshot.digest,
                config_files.digest,
                settings,
                ignore_files.digest,
            )
        )
    )

    cache_scope = ProcessCacheScope.PER_SESSION if semgrep.force else ProcessCacheScope.SUCCESSFUL

    # TODO: https://github.com/pantsbuild/pants/issues/18430 support running this with --autofix
    # under the fix goal... but not all rules have fixes, so we need to be running with
    # --error/checking exit codes, which FixResult doesn't currently support.
    result = await execute_process(
        **implicitly(
            VenvPexProcess(
                semgrep_pex,
                argv=(
                    "scan",
                    *(f"--config={f}" for f in config_files.files),
                    "--jobs={pants_concurrency}",
                    "--error",
                    *semgrep.args,
                    # we don't pass the target files directly because that overrides .semgrepignore
                    # (https://github.com/returntocorp/semgrep/issues/4978), so instead we just tell its
                    # traversal to include all the source files in this partition. Unfortunately this
                    # include is implicitly unrooted (i.e. as if it was **/path/to/file), and so may
                    # pick up other files if the names match. The highest risk of this is within the
                    # semgrep PEX.
                    *(f"--include={f}" for f in input_files.files),
                    f"--exclude={semgrep_pex.pex_filename}",
                ),
                extra_env={
                    "SEMGREP_FORCE_COLOR": "true",
                    # disable various global state/network requests
                    "SEMGREP_SETTINGS_FILE": _DEFAULT_SETTINGS.path,
                    "SEMGREP_ENABLE_VERSION_CHECK": "0",
                    "SEMGREP_SEND_METRICS": "off",
                },
                input_digest=input_digest,
                concurrency_available=len(input_files.files),
                description=f"Run Semgrep on {pluralize(len(input_files.files), 'file')}.",
                level=LogLevel.DEBUG,
                cache_scope=cache_scope,
            )
        )
    )

    return LintResult.create(request, result, output_simplifier=global_options.output_simplifier())


def rules() -> Iterable[Rule | UnionRule]:
    return [*collect_rules(), *SemgrepLintRequest.rules(), *pex.rules()]

1	# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	1✔
4
5	import itertools	1✔
6	import logging	1✔
7	from collections import defaultdict	1✔
8	from collections.abc import Iterable	1✔
9	from dataclasses import dataclass	1✔
10	from pathlib import PurePath	1✔
11
12	from pants.backend.python.util_rules import pex	1✔
13	from pants.backend.python.util_rules.pex import VenvPexProcess, create_venv_pex	1✔
14	from pants.core.goals.lint import LintResult, LintTargetsRequest	1✔
15	from pants.core.util_rules.partitions import Partition, Partitions	1✔
16	from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files	1✔
17	from pants.engine.addresses import Address	1✔
18	from pants.engine.fs import CreateDigest, FileContent, MergeDigests, PathGlobs, Paths	1✔
19	from pants.engine.intrinsics import (	1✔
20	create_digest,
21	digest_to_snapshot,
22	execute_process,
23	merge_digests,
24	path_globs_to_paths,
25	)
26	from pants.engine.process import ProcessCacheScope	1✔
27	from pants.engine.rules import Rule, collect_rules, concurrently, implicitly, rule	1✔
28	from pants.engine.unions import UnionRule	1✔
29	from pants.option.global_options import GlobalOptions	1✔
30	from pants.util.logging import LogLevel	1✔
31	from pants.util.strutil import pluralize	1✔
32
33	from .subsystem import SemgrepFieldSet, SemgrepSubsystem	1✔
34
35	logger = logging.getLogger(__name__)	1✔
36
37
38	_SEMGREPIGNORE_FILE_NAME = ".semgrepignore"	1✔
39	_DEFAULT_SEMGREP_CONFIG_DIR = ".semgrep"	1✔
40
41
42	class SemgrepLintRequest(LintTargetsRequest):	1✔
43	field_set_type = SemgrepFieldSet	1✔
44	tool_subsystem = SemgrepSubsystem # type: ignore[assignment]	1✔
45
46
47	@dataclass(frozen=True)	1✔
48	class PartitionMetadata:	1✔
49	config_files: frozenset[PurePath]	1✔
50
51	@property	1✔
52	def description(self) -> str:	1✔
53	return ", ".join(sorted(str(path) for path in self.config_files))	×
54
55
56	@dataclass	1✔
57	class AllSemgrepConfigs:	1✔
58	configs_by_dir: dict[PurePath, set[PurePath]]	1✔
59
60	def ancestor_configs(self, address: Address) -> Iterable[PurePath]:	1✔
61	# TODO: introspect the semgrep rules and determine which (if any) apply to the files, e.g. a
62	# Python file shouldn't depend on a .semgrep.yml that doesn't have any 'python' or 'generic'
63	# rules, and similarly if there's path inclusions/exclusions.
64	# TODO: this would be better as actual dependency inference (e.g. allows inspection, manual
65	# addition/exclusion), but that can only infer 'full' dependencies and it is wrong (e.g. JVM
66	# things break) for real code files to depend on this sort of non-code linter config; requires
67	# dependency scopes or similar (https://github.com/pantsbuild/pants/issues/12794)
UNCOV 68	spec = PurePath(address.spec_path)	×
69
UNCOV 70	for ancestor in itertools.chain([spec], spec.parents):	×
UNCOV 71	yield from self.configs_by_dir.get(ancestor, [])	×
72
73
74	def _group_by_semgrep_dir(	1✔
75	all_config_files: Paths, all_config_dir_files: Paths, config_name: str
76	) -> AllSemgrepConfigs:
UNCOV 77	configs_by_dir: dict[PurePath, set[PurePath]] = {}	×
UNCOV 78	for config_path in all_config_files.files:	×
79	# Rules like foo/semgrep.yaml should apply to the project at foo/
UNCOV 80	path = PurePath(config_path)	×
UNCOV 81	configs_by_dir.setdefault(path.parent, set()).add(path)	×
82
UNCOV 83	for config_path in all_config_dir_files.files:	×
84	# Rules like foo/bar/.semgrep/baz.yaml and foo/bar/.semgrep/baz/qux.yaml should apply to the
85	# project at foo/bar/
UNCOV 86	path = PurePath(config_path)	×
UNCOV 87	config_directory = next(	×
88	parent.parent for parent in path.parents if parent.name == config_name
89	)
UNCOV 90	configs_by_dir.setdefault(config_directory, set()).add(path)	×
91
UNCOV 92	return AllSemgrepConfigs(configs_by_dir)	×
93
94
95	@rule	1✔
96	async def find_all_semgrep_configs(semgrep: SemgrepSubsystem) -> AllSemgrepConfigs:	1✔
97	config_file_globs: tuple[str, ...] = ()	×
98	config_dir_globs: tuple[str, ...] = ()	×
99
100	if semgrep.config_name is None:	×
101	config_file_globs = ("/.semgrep.yml", "/.semgrep.yaml")	×
102	config_dir_globs = (	×
103	f"/{_DEFAULT_SEMGREP_CONFIG_DIR}//*.yaml",
104	f"/{_DEFAULT_SEMGREP_CONFIG_DIR}//*.yml",
105	)
106	elif semgrep.config_name.endswith((".yaml", ".yml")):	×
107	config_file_globs = (f"**/{semgrep.config_name}",)	×
108	else:
109	config_dir_globs = (	×
110	f"/{semgrep.config_name}//*.yaml",
111	f"/{semgrep.config_name}//*.yml",
112	)
113
114	all_config_files = await path_globs_to_paths(PathGlobs(config_file_globs))	×
115	all_config_dir_files = await path_globs_to_paths(PathGlobs(config_dir_globs))	×
116	return _group_by_semgrep_dir(	×
117	all_config_files,
118	all_config_dir_files,
119	(semgrep.config_name or _DEFAULT_SEMGREP_CONFIG_DIR),
120	)
121
122
123	@dataclass(frozen=True)	1✔
124	class RelevantSemgrepConfigsRequest:	1✔
125	field_set: SemgrepFieldSet	1✔
126
127
128	class RelevantSemgrepConfigs(frozenset[PurePath]):	1✔
129	pass	1✔
130
131
132	@rule	1✔
133	async def infer_relevant_semgrep_configs(	1✔
134	request: RelevantSemgrepConfigsRequest, all_semgrep: AllSemgrepConfigs
135	) -> RelevantSemgrepConfigs:
136	return RelevantSemgrepConfigs(all_semgrep.ancestor_configs(request.field_set.address))	×
137
138
139	@rule	1✔
140	async def partition(	1✔
141	request: SemgrepLintRequest.PartitionRequest[SemgrepFieldSet],
142	semgrep: SemgrepSubsystem,
143	) -> Partitions:
144	if semgrep.skip:	×
145	return Partitions()	×
146
147	all_configs = await concurrently(	×
148	infer_relevant_semgrep_configs(RelevantSemgrepConfigsRequest(field_set), **implicitly())
149	for field_set in request.field_sets
150	)
151
152	# partition by the sets of configs that apply to each input
153	by_config = defaultdict(list)	×
154	for field_set, configs in zip(request.field_sets, all_configs):	×
155	if configs:	×
156	by_config[configs].append(field_set)	×
157
158	return Partitions(	×
159	Partition(tuple(field_sets), PartitionMetadata(configs))
160	for configs, field_sets in by_config.items()
161	)
162
163
164	# We have a hard-coded settings file to side-step
165	# https://github.com/returntocorp/semgrep/issues/7102, and also provide more cacheability, NB. both
166	# keys are required.
167	_DEFAULT_SETTINGS = FileContent(	1✔
168	path="__semgrep_settings.yaml",
169	content=b"anonymous_user_id: 00000000-0000-0000-0000-000000000000\nhas_shown_metrics_notification: true",
170	)
171
172
173	@rule(desc="Lint with Semgrep", level=LogLevel.DEBUG)	1✔
174	async def lint(	1✔
175	request: SemgrepLintRequest.Batch[SemgrepFieldSet, PartitionMetadata],
176	semgrep: SemgrepSubsystem,
177	global_options: GlobalOptions,
178	) -> LintResult:
179	config_files, ignore_files, semgrep_pex, input_files, settings = await concurrently(	×
180	digest_to_snapshot(
181	**implicitly(PathGlobs(str(s) for s in request.partition_metadata.config_files))
182	),
183	digest_to_snapshot(**implicitly(PathGlobs([_SEMGREPIGNORE_FILE_NAME]))),
184	create_venv_pex(**implicitly(semgrep.to_pex_request())),
185	determine_source_files(
186	SourceFilesRequest(field_set.source for field_set in request.elements)
187	),
188	create_digest(CreateDigest([_DEFAULT_SETTINGS])),
189	)
190
191	input_digest = await merge_digests(	×
192	MergeDigests(
193	(
194	input_files.snapshot.digest,
195	config_files.digest,
196	settings,
197	ignore_files.digest,
198	)
199	)
200	)
201
202	cache_scope = ProcessCacheScope.PER_SESSION if semgrep.force else ProcessCacheScope.SUCCESSFUL	×
203
204	# TODO: https://github.com/pantsbuild/pants/issues/18430 support running this with --autofix
205	# under the fix goal... but not all rules have fixes, so we need to be running with
206	# --error/checking exit codes, which FixResult doesn't currently support.
207	result = await execute_process(	×
208	**implicitly(
209	VenvPexProcess(
210	semgrep_pex,
211	argv=(
212	"scan",
213	*(f"--config={f}" for f in config_files.files),
214	"--jobs={pants_concurrency}",
215	"--error",
216	*semgrep.args,
217	# we don't pass the target files directly because that overrides .semgrepignore
218	# (https://github.com/returntocorp/semgrep/issues/4978), so instead we just tell its
219	# traversal to include all the source files in this partition. Unfortunately this
220	# include is implicitly unrooted (i.e. as if it was **/path/to/file), and so may
221	# pick up other files if the names match. The highest risk of this is within the
222	# semgrep PEX.
223	*(f"--include={f}" for f in input_files.files),
224	f"--exclude={semgrep_pex.pex_filename}",
225	),
226	extra_env={
227	"SEMGREP_FORCE_COLOR": "true",
228	# disable various global state/network requests
229	"SEMGREP_SETTINGS_FILE": _DEFAULT_SETTINGS.path,
230	"SEMGREP_ENABLE_VERSION_CHECK": "0",
231	"SEMGREP_SEND_METRICS": "off",
232	},
233	input_digest=input_digest,
234	concurrency_available=len(input_files.files),
235	description=f"Run Semgrep on {pluralize(len(input_files.files), 'file')}.",
236	level=LogLevel.DEBUG,
237	cache_scope=cache_scope,
238	)
239	)
240	)
241
242	return LintResult.create(request, result, output_simplifier=global_options.output_simplifier())	×
243
244
245	def rules() -> Iterable[Rule \| UnionRule]:	1✔
246	return [collect_rules(), SemgrepLintRequest.rules(), *pex.rules()]	1✔

pantsbuild / pants / 19381742489

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous