25441711719

Committed 06 May 2026 02:31PM UTC coverage: 92.915%. Remained the same

Build # 25441711719

Build Type

push

github

Committed by

web-flow

Commit Message

use sha pin (with comment) format for generated actions (#23312)

Per the GitHub Action best practices we recently enabled at #23249, we
should pin each action to a SHA so that the reference is actually
immutable.

This will -- I hope -- knock out a large chunk of the 421 alerts we
currently get from zizmor. The next followup would then be upgrades and
harmonizing the generated and none-generated pins.

Notice: This idea was suggested by Claude while going over pinact output
and I was surprised to see that post processing the yaml wasn't too
gross.

Coverage Stats

92206 of 99237 relevant lines covered (92.91%)

4.04 hits per line

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

96.43

/src/python/pants/backend/docker/util_rules/docker_binary.py

# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import logging
import os
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import cast

from pants.backend.docker.subsystems.docker_options import DockerOptions
from pants.backend.docker.util_rules.docker_build_args import DockerBuildArgs
from pants.core.util_rules.system_binaries import (
    BinaryPath,
    BinaryPathRequest,
    BinaryPathTest,
    BinaryShims,
    BinaryShimsRequest,
    create_binary_shims,
    find_binary,
)
from pants.engine.fs import Digest
from pants.engine.internals.selectors import concurrently
from pants.engine.process import Process, ProcessCacheScope
from pants.engine.rules import collect_rules, implicitly, rule
from pants.util.logging import LogLevel
from pants.util.strutil import pluralize

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class DockerBinary(BinaryPath):
    """The `docker` binary."""

    extra_env: Mapping[str, str]
    extra_input_digests: Mapping[str, Digest] | None

    is_podman: bool

    def __init__(
        self,
        path: str,
        fingerprint: str | None = None,
        extra_env: Mapping[str, str] | None = None,
        extra_input_digests: Mapping[str, Digest] | None = None,
        is_podman: bool = False,
    ) -> None:
        object.__setattr__(self, "extra_env", {} if extra_env is None else extra_env)
        object.__setattr__(self, "extra_input_digests", extra_input_digests)
        object.__setattr__(self, "is_podman", is_podman)
        super().__init__(path, fingerprint)

    def _get_process_environment(self, env: Mapping[str, str]) -> Mapping[str, str]:
        if not self.extra_env:
            return env

        res = {**self.extra_env, **env}

        # Merge the PATH entries, in case they are present in both `env` and `self.extra_env`.
        res["PATH"] = os.pathsep.join(
            p for p in (m.get("PATH") for m in (self.extra_env, env)) if p
        )
        return res

    def build_image(
        self,
        tags: tuple[str, ...],
        digest: Digest,
        dockerfile: str,
        build_args: DockerBuildArgs,
        context_root: str,
        env: Mapping[str, str],
        use_buildx: bool,
        extra_args: tuple[str, ...] = (),
        output_files: tuple[str, ...] = (),
        output_directories: tuple[str, ...] = (),
    ) -> Process:
        if use_buildx:
            build_commands = ["buildx", "build"]
        else:
            build_commands = ["build"]

        args = [self.path, *build_commands, *extra_args]

        for tag in tags:
            args.extend(["--tag", tag])

        for build_arg in build_args:
            args.extend(["--build-arg", build_arg])

        args.extend(["--file", dockerfile])

        # Docker context root.
        args.append(context_root)

        return Process(
            argv=tuple(args),
            description=(
                f"Building docker image {tags[0]}"
                + (f" +{pluralize(len(tags) - 1, 'additional tag')}." if len(tags) > 1 else "")
            ),
            env=self._get_process_environment(env),
            input_digest=digest,
            immutable_input_digests=self.extra_input_digests,
            output_files=output_files,
            output_directories=output_directories,
            # We must run the docker build commands every time, even if nothing has changed,
            # in case the user ran `docker image rm` outside of Pants.
            cache_scope=ProcessCacheScope.PER_SESSION,
        )

    def push_image(self, tag: str, env: Mapping[str, str] | None = None) -> Process:
        return Process(
            argv=(self.path, "push", tag),
            cache_scope=ProcessCacheScope.PER_SESSION,
            description=f"Pushing docker image {tag}",
            env=self._get_process_environment(env or {}),
            immutable_input_digests=self.extra_input_digests,
        )

    def run_image(
        self,
        tag: str,
        *,
        docker_run_args: tuple[str, ...] | None = None,
        image_args: tuple[str, ...] | None = None,
        env: Mapping[str, str] | None = None,
    ) -> Process:
        return Process(
            argv=(self.path, "run", *(docker_run_args or []), tag, *(image_args or [])),
            cache_scope=ProcessCacheScope.PER_SESSION,
            description=f"Running docker image {tag}",
            env=self._get_process_environment(env or {}),
            immutable_input_digests=self.extra_input_digests,
        )


async def _get_docker_tools_shims(
    *,
    tools: Sequence[str],
    optional_tools: Sequence[str],
    search_path: Sequence[str],
    rationale: str,
) -> BinaryShims:
    all_binary_first_paths: list[BinaryPath] = []

    if tools:
        tools_requests = [
            BinaryPathRequest(binary_name=binary_name, search_path=search_path)
            for binary_name in tools
        ]

        tools_paths = await concurrently(
            find_binary(tools_request, **implicitly()) for tools_request in tools_requests
        )

        all_binary_first_paths.extend(
            [
                path.first_path_or_raise(request, rationale=rationale)
                for request, path in zip(tools_requests, tools_paths)
            ]
        )

    if optional_tools:
        optional_tools_requests = [
            BinaryPathRequest(binary_name=binary_name, search_path=search_path)
            for binary_name in optional_tools
        ]

        optional_tools_paths = await concurrently(
            find_binary(optional_tools_request, **implicitly())
            for optional_tools_request in optional_tools_requests
        )

        all_binary_first_paths.extend(
            [
                cast(BinaryPath, path.first_path)  # safe since we check for non-empty paths below
                for path in optional_tools_paths
                if path.paths
            ]
        )

    tools_shims = await create_binary_shims(
        BinaryShimsRequest.for_paths(
            *all_binary_first_paths,
            rationale=rationale,
        ),
        **implicitly(),
    )

    return tools_shims


@rule(desc="Finding the `docker` binary and related tooling", level=LogLevel.DEBUG)
async def get_docker(
    docker_options: DockerOptions, docker_options_env_aware: DockerOptions.EnvironmentAware
) -> DockerBinary:
    search_path = docker_options_env_aware.executable_search_path

    first_path: BinaryPath | None = None
    is_podman = False

    if getattr(docker_options.options, "experimental_enable_podman", False):
        # Enable podman support with `pants.backend.experimental.docker.podman`
        request = BinaryPathRequest(
            binary_name="podman",
            search_path=search_path,
            test=BinaryPathTest(args=["-v"]),
        )
        paths = await find_binary(request, **implicitly())
        first_path = paths.first_path
        if first_path:
            is_podman = True
            logger.warning("podman found. Podman support is experimental.")

    if not first_path:
        request = BinaryPathRequest(
            binary_name="docker",
            search_path=search_path,
            test=BinaryPathTest(args=["-v"]),
        )
        paths = await find_binary(request, **implicitly())
        first_path = paths.first_path_or_raise(request, rationale="interact with the docker daemon")

    if not docker_options.tools and not docker_options.optional_tools:
        return DockerBinary(first_path.path, first_path.fingerprint, is_podman=is_podman)

    tools_shims = await _get_docker_tools_shims(
        tools=docker_options.tools,
        optional_tools=docker_options.optional_tools,
        search_path=search_path,
        rationale="use docker",
    )

    extra_env = {"PATH": tools_shims.path_component}
    extra_input_digests = tools_shims.immutable_input_digests

    return DockerBinary(
        first_path.path,
        first_path.fingerprint,
        extra_env=extra_env,
        extra_input_digests=extra_input_digests,
        is_podman=is_podman,
    )


def rules():
    return collect_rules()

1	# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	9✔
5
6	import logging	9✔
7	import os	9✔
8	from collections.abc import Mapping, Sequence	9✔
9	from dataclasses import dataclass	9✔
10	from typing import cast	9✔
11
12	from pants.backend.docker.subsystems.docker_options import DockerOptions	9✔
13	from pants.backend.docker.util_rules.docker_build_args import DockerBuildArgs	9✔
14	from pants.core.util_rules.system_binaries import (	9✔
15	BinaryPath,
16	BinaryPathRequest,
17	BinaryPathTest,
18	BinaryShims,
19	BinaryShimsRequest,
20	create_binary_shims,
21	find_binary,
22	)
23	from pants.engine.fs import Digest	9✔
24	from pants.engine.internals.selectors import concurrently	9✔
25	from pants.engine.process import Process, ProcessCacheScope	9✔
26	from pants.engine.rules import collect_rules, implicitly, rule	9✔
27	from pants.util.logging import LogLevel	9✔
28	from pants.util.strutil import pluralize	9✔
29
30	logger = logging.getLogger(__name__)	9✔
31
32
33	@dataclass(frozen=True)	9✔
34	class DockerBinary(BinaryPath):	9✔
35	"""The `docker` binary."""
36
37	extra_env: Mapping[str, str]	9✔
38	extra_input_digests: Mapping[str, Digest] \| None	9✔
39
40	is_podman: bool	9✔
41
42	def __init__(	9✔
43	self,
44	path: str,
45	fingerprint: str \| None = None,
46	extra_env: Mapping[str, str] \| None = None,
47	extra_input_digests: Mapping[str, Digest] \| None = None,
48	is_podman: bool = False,
49	) -> None:
50	object.__setattr__(self, "extra_env", {} if extra_env is None else extra_env)	5✔
51	object.__setattr__(self, "extra_input_digests", extra_input_digests)	5✔
52	object.__setattr__(self, "is_podman", is_podman)	5✔
53	super().__init__(path, fingerprint)	5✔
54
55	def _get_process_environment(self, env: Mapping[str, str]) -> Mapping[str, str]:	9✔
56	if not self.extra_env:	5✔
57	return env	5✔
58
59	res = {self.extra_env, env}	×
60
61	# Merge the PATH entries, in case they are present in both `env` and `self.extra_env`.
62	res["PATH"] = os.pathsep.join(	×
63	p for p in (m.get("PATH") for m in (self.extra_env, env)) if p
64	)
65	return res	×
66
67	def build_image(	9✔
68	self,
69	tags: tuple[str, ...],
70	digest: Digest,
71	dockerfile: str,
72	build_args: DockerBuildArgs,
73	context_root: str,
74	env: Mapping[str, str],
75	use_buildx: bool,
76	extra_args: tuple[str, ...] = (),
77	output_files: tuple[str, ...] = (),
78	output_directories: tuple[str, ...] = (),
79	) -> Process:
80	if use_buildx:	4✔
81	build_commands = ["buildx", "build"]	1✔
82	else:
83	build_commands = ["build"]	4✔
84
85	args = [self.path, build_commands, extra_args]	4✔
86
87	for tag in tags:	4✔
88	args.extend(["--tag", tag])	4✔
89
90	for build_arg in build_args:	4✔
91	args.extend(["--build-arg", build_arg])	2✔
92
93	args.extend(["--file", dockerfile])	4✔
94
95	# Docker context root.
96	args.append(context_root)	4✔
97
98	return Process(	4✔
99	argv=tuple(args),
100	description=(
101	f"Building docker image {tags[0]}"
102	+ (f" +{pluralize(len(tags) - 1, 'additional tag')}." if len(tags) > 1 else "")
103	),
104	env=self._get_process_environment(env),
105	input_digest=digest,
106	immutable_input_digests=self.extra_input_digests,
107	output_files=output_files,
108	output_directories=output_directories,
109	# We must run the docker build commands every time, even if nothing has changed,
110	# in case the user ran `docker image rm` outside of Pants.
111	cache_scope=ProcessCacheScope.PER_SESSION,
112	)
113
114	def push_image(self, tag: str, env: Mapping[str, str] \| None = None) -> Process:	9✔
115	return Process(	2✔
116	argv=(self.path, "push", tag),
117	cache_scope=ProcessCacheScope.PER_SESSION,
118	description=f"Pushing docker image {tag}",
119	env=self._get_process_environment(env or {}),
120	immutable_input_digests=self.extra_input_digests,
121	)
122
123	def run_image(	9✔
124	self,
125	tag: str,
126	*,
127	docker_run_args: tuple[str, ...] \| None = None,
128	image_args: tuple[str, ...] \| None = None,
129	env: Mapping[str, str] \| None = None,
130	) -> Process:
131	return Process(	1✔
132	argv=(self.path, "run", (docker_run_args or []), tag, (image_args or [])),
133	cache_scope=ProcessCacheScope.PER_SESSION,
134	description=f"Running docker image {tag}",
135	env=self._get_process_environment(env or {}),
136	immutable_input_digests=self.extra_input_digests,
137	)
138
139
140	async def _get_docker_tools_shims(	9✔
141	*,
142	tools: Sequence[str],
143	optional_tools: Sequence[str],
144	search_path: Sequence[str],
145	rationale: str,
146	) -> BinaryShims:
147	all_binary_first_paths: list[BinaryPath] = []	1✔
148
149	if tools:	1✔
150	tools_requests = [	1✔
151	BinaryPathRequest(binary_name=binary_name, search_path=search_path)
152	for binary_name in tools
153	]
154
155	tools_paths = await concurrently(	1✔
156	find_binary(tools_request, **implicitly()) for tools_request in tools_requests
157	)
158
159	all_binary_first_paths.extend(	1✔
160	[
161	path.first_path_or_raise(request, rationale=rationale)
162	for request, path in zip(tools_requests, tools_paths)
163	]
164	)
165
166	if optional_tools:	1✔
167	optional_tools_requests = [	1✔
168	BinaryPathRequest(binary_name=binary_name, search_path=search_path)
169	for binary_name in optional_tools
170	]
171
172	optional_tools_paths = await concurrently(	1✔
173	find_binary(optional_tools_request, **implicitly())
174	for optional_tools_request in optional_tools_requests
175	)
176
177	all_binary_first_paths.extend(	1✔
178	[
179	cast(BinaryPath, path.first_path) # safe since we check for non-empty paths below
180	for path in optional_tools_paths
181	if path.paths
182	]
183	)
184
185	tools_shims = await create_binary_shims(	1✔
186	BinaryShimsRequest.for_paths(
187	*all_binary_first_paths,
188	rationale=rationale,
189	),
190	**implicitly(),
191	)
192
193	return tools_shims	1✔
194
195
196	@rule(desc="Finding the `docker` binary and related tooling", level=LogLevel.DEBUG)	9✔
197	async def get_docker(	9✔
198	docker_options: DockerOptions, docker_options_env_aware: DockerOptions.EnvironmentAware
199	) -> DockerBinary:
200	search_path = docker_options_env_aware.executable_search_path	4✔
201
202	first_path: BinaryPath \| None = None	4✔
203	is_podman = False	4✔
204
205	if getattr(docker_options.options, "experimental_enable_podman", False):	4✔
206	# Enable podman support with `pants.backend.experimental.docker.podman`
207	request = BinaryPathRequest(	1✔
208	binary_name="podman",
209	search_path=search_path,
210	test=BinaryPathTest(args=["-v"]),
211	)
212	paths = await find_binary(request, **implicitly())	1✔
213	first_path = paths.first_path	1✔
214	if first_path:	1✔
215	is_podman = True	1✔
216	logger.warning("podman found. Podman support is experimental.")	1✔
217
218	if not first_path:	4✔
219	request = BinaryPathRequest(	4✔
220	binary_name="docker",
221	search_path=search_path,
222	test=BinaryPathTest(args=["-v"]),
223	)
224	paths = await find_binary(request, **implicitly())	4✔
225	first_path = paths.first_path_or_raise(request, rationale="interact with the docker daemon")	4✔
226
227	if not docker_options.tools and not docker_options.optional_tools:	4✔
228	return DockerBinary(first_path.path, first_path.fingerprint, is_podman=is_podman)	4✔
229
230	tools_shims = await _get_docker_tools_shims(	1✔
231	tools=docker_options.tools,
232	optional_tools=docker_options.optional_tools,
233	search_path=search_path,
234	rationale="use docker",
235	)
236
237	extra_env = {"PATH": tools_shims.path_component}	1✔
238	extra_input_digests = tools_shims.immutable_input_digests	1✔
239
240	return DockerBinary(	1✔
241	first_path.path,
242	first_path.fingerprint,
243	extra_env=extra_env,
244	extra_input_digests=extra_input_digests,
245	is_podman=is_podman,
246	)
247
248
249	def rules():	9✔
250	return collect_rules()	9✔

pantsbuild / pants / 25441711719

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