We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Pants `export` doesn't work on all machines
**Describe the bug**
I am able to run `./pants export ::` successfully on my machine. When I ran this through our CI system, I got a permission denied error when trying to clone a VCS requirement from our requirements.txt file. I know the CI worker had access to the repos because it cloned a repo to get this far (the VCS requirement is another internal repo).

I already have the following added to `pants.toml`.
```
[subprocess-environment]
env_vars.add = ["SSH_AUTH_SOCK"]
```

I'm curious if it's possible this isn't getting passed through on my CI system and causing the export step to fail.

**Pants version**
2.11.0

**OS**
ubuntu 20.04



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/python/pants/backend/python/util_rules/pex_cli.py`
Content:
```
1 # Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2 # Licensed under the Apache License, Version 2.0 (see LICENSE).
3 
4 from __future__ import annotations
5 
6 import dataclasses
7 from dataclasses import dataclass
8 from typing import Iterable, List, Mapping, Optional, Tuple
9 
10 from pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem
11 from pants.backend.python.util_rules import pex_environment
12 from pants.backend.python.util_rules.pex_environment import (
13     PexEnvironment,
14     PexSubsystem,
15     PythonExecutable,
16 )
17 from pants.core.util_rules import external_tool
18 from pants.core.util_rules.external_tool import (
19     DownloadedExternalTool,
20     ExternalToolRequest,
21     TemplatedExternalTool,
22 )
23 from pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests
24 from pants.engine.internals.selectors import MultiGet
25 from pants.engine.platform import Platform
26 from pants.engine.process import Process, ProcessCacheScope
27 from pants.engine.rules import Get, collect_rules, rule
28 from pants.option.global_options import GlobalOptions, ca_certs_path_to_file_content
29 from pants.util.frozendict import FrozenDict
30 from pants.util.logging import LogLevel
31 from pants.util.meta import classproperty, frozen_after_init
32 from pants.util.strutil import create_path_env_var
33 
34 
35 class PexCli(TemplatedExternalTool):
36     options_scope = "pex-cli"
37     name = "pex"
38     help = "The PEX (Python EXecutable) tool (https://github.com/pantsbuild/pex)."
39 
40     default_version = "v2.1.103"
41     default_url_template = "https://github.com/pantsbuild/pex/releases/download/{version}/pex"
42     version_constraints = ">=2.1.103,<3.0"
43 
44     @classproperty
45     def default_known_versions(cls):
46         return [
47             "|".join(
48                 (
49                     cls.default_version,
50                     plat,
51                     "4d45336511484100ae4e2bab24542a8b86b12c8cb89230463593c60d08c4b8d3",
52                     "3814407",
53                 )
54             )
55             for plat in ["macos_arm64", "macos_x86_64", "linux_x86_64", "linux_arm64"]
56         ]
57 
58 
59 @frozen_after_init
60 @dataclass(unsafe_hash=True)
61 class PexCliProcess:
62     subcommand: tuple[str, ...]
63     extra_args: tuple[str, ...]
64     set_resolve_args: bool
65     description: str = dataclasses.field(compare=False)
66     additional_input_digest: Optional[Digest]
67     extra_env: Optional[FrozenDict[str, str]]
68     output_files: Optional[Tuple[str, ...]]
69     output_directories: Optional[Tuple[str, ...]]
70     python: Optional[PythonExecutable]
71     level: LogLevel
72     concurrency_available: int
73     cache_scope: ProcessCacheScope
74 
75     def __init__(
76         self,
77         *,
78         subcommand: Iterable[str],
79         extra_args: Iterable[str],
80         description: str,
81         set_resolve_args: bool = True,
82         additional_input_digest: Optional[Digest] = None,
83         extra_env: Optional[Mapping[str, str]] = None,
84         output_files: Optional[Iterable[str]] = None,
85         output_directories: Optional[Iterable[str]] = None,
86         python: Optional[PythonExecutable] = None,
87         level: LogLevel = LogLevel.INFO,
88         concurrency_available: int = 0,
89         cache_scope: ProcessCacheScope = ProcessCacheScope.SUCCESSFUL,
90     ) -> None:
91         self.subcommand = tuple(subcommand)
92         self.extra_args = tuple(extra_args)
93         self.set_resolve_args = set_resolve_args
94         self.description = description
95         self.additional_input_digest = additional_input_digest
96         self.extra_env = FrozenDict(extra_env) if extra_env else None
97         self.output_files = tuple(output_files) if output_files else None
98         self.output_directories = tuple(output_directories) if output_directories else None
99         self.python = python
100         self.level = level
101         self.concurrency_available = concurrency_available
102         self.cache_scope = cache_scope
103         self.__post_init__()
104 
105     def __post_init__(self) -> None:
106         if "--pex-root-path" in self.extra_args:
107             raise ValueError("`--pex-root` flag not allowed. We set its value for you.")
108 
109 
110 class PexPEX(DownloadedExternalTool):
111     """The Pex PEX binary."""
112 
113 
114 @rule
115 async def download_pex_pex(pex_cli: PexCli, platform: Platform) -> PexPEX:
116     pex_pex = await Get(DownloadedExternalTool, ExternalToolRequest, pex_cli.get_request(platform))
117     return PexPEX(digest=pex_pex.digest, exe=pex_pex.exe)
118 
119 
120 @rule
121 async def setup_pex_cli_process(
122     request: PexCliProcess,
123     pex_pex: PexPEX,
124     pex_env: PexEnvironment,
125     python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,
126     global_options: GlobalOptions,
127     pex_subsystem: PexSubsystem,
128 ) -> Process:
129     tmpdir = ".tmp"
130     gets: List[Get] = [Get(Digest, CreateDigest([Directory(tmpdir)]))]
131 
132     cert_args = []
133     if global_options.ca_certs_path:
134         ca_certs_fc = ca_certs_path_to_file_content(global_options.ca_certs_path)
135         gets.append(Get(Digest, CreateDigest((ca_certs_fc,))))
136         cert_args = ["--cert", ca_certs_fc.path]
137 
138     digests_to_merge = [pex_pex.digest]
139     digests_to_merge.extend(await MultiGet(gets))
140     if request.additional_input_digest:
141         digests_to_merge.append(request.additional_input_digest)
142     input_digest = await Get(Digest, MergeDigests(digests_to_merge))
143 
144     global_args = [
145         # Ensure Pex and its subprocesses create temporary files in the the process execution
146         # sandbox. It may make sense to do this generally for Processes, but in the short term we
147         # have known use cases where /tmp is too small to hold large wheel downloads Pex is asked to
148         # perform. Making the TMPDIR local to the sandbox allows control via
149         # --local-execution-root-dir for the local case and should work well with remote cases where
150         # a remoting implementation has to allow for processes producing large binaries in a
151         # sandbox to support reasonable workloads. Communicating TMPDIR via --tmpdir instead of via
152         # environment variable allows Pex to absolutize the path ensuring subprocesses that change
153         # CWD can find the TMPDIR.
154         "--tmpdir",
155         tmpdir,
156     ]
157 
158     if request.concurrency_available > 0:
159         global_args.extend(["--jobs", "{pants_concurrency}"])
160 
161     verbosity_args = [f"-{'v' * pex_subsystem.verbosity}"] if pex_subsystem.verbosity > 0 else []
162 
163     resolve_args = (
164         [*cert_args, "--python-path", create_path_env_var(pex_env.interpreter_search_paths)]
165         if request.set_resolve_args
166         else []
167     )
168     args = [
169         *request.subcommand,
170         *global_args,
171         *verbosity_args,
172         *resolve_args,
173         # NB: This comes at the end because it may use `--` passthrough args, # which must come at
174         # the end.
175         *request.extra_args,
176     ]
177 
178     complete_pex_env = pex_env.in_sandbox(working_directory=None)
179     normalized_argv = complete_pex_env.create_argv(pex_pex.exe, *args, python=request.python)
180     env = {
181         **complete_pex_env.environment_dict(python_configured=request.python is not None),
182         **python_native_code.subprocess_env_vars,
183         **(request.extra_env or {}),
184         # If a subcommand is used, we need to use the `pex3` console script.
185         **({"PEX_SCRIPT": "pex3"} if request.subcommand else {}),
186     }
187 
188     return Process(
189         normalized_argv,
190         description=request.description,
191         input_digest=input_digest,
192         env=env,
193         output_files=request.output_files,
194         output_directories=request.output_directories,
195         append_only_caches=complete_pex_env.append_only_caches,
196         level=request.level,
197         concurrency_available=request.concurrency_available,
198         cache_scope=request.cache_scope,
199     )
200 
201 
202 def rules():
203     return [
204         *collect_rules(),
205         *external_tool.rules(),
206         *pex_environment.rules(),
207     ]
208
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/python/pants/backend/python/util_rules/pex_cli.py b/src/python/pants/backend/python/util_rules/pex_cli.py
--- a/src/python/pants/backend/python/util_rules/pex_cli.py
+++ b/src/python/pants/backend/python/util_rules/pex_cli.py
@@ -37,7 +37,7 @@
     name = "pex"
     help = "The PEX (Python EXecutable) tool (https://github.com/pantsbuild/pex)."
 
-    default_version = "v2.1.103"
+    default_version = "v2.1.107"
     default_url_template = "https://github.com/pantsbuild/pex/releases/download/{version}/pex"
     version_constraints = ">=2.1.103,<3.0"
 
@@ -48,8 +48,8 @@
                 (
                     cls.default_version,
                     plat,
-                    "4d45336511484100ae4e2bab24542a8b86b12c8cb89230463593c60d08c4b8d3",
-                    "3814407",
+                    "bfc19b16e0f298742dd933289bd8057dd503f9ad0678310412d382800d48b3ae",
+                    "3840814",
                 )
             )
             for plat in ["macos_arm64", "macos_x86_64", "linux_x86_64", "linux_arm64"]

{"golden_diff": "diff --git a/src/python/pants/backend/python/util_rules/pex_cli.py b/src/python/pants/backend/python/util_rules/pex_cli.py\n--- a/src/python/pants/backend/python/util_rules/pex_cli.py\n+++ b/src/python/pants/backend/python/util_rules/pex_cli.py\n@@ -37,7 +37,7 @@\n     name = \"pex\"\n     help = \"The PEX (Python EXecutable) tool (https://github.com/pantsbuild/pex).\"\n \n-    default_version = \"v2.1.103\"\n+    default_version = \"v2.1.107\"\n     default_url_template = \"https://github.com/pantsbuild/pex/releases/download/{version}/pex\"\n     version_constraints = \">=2.1.103,<3.0\"\n \n@@ -48,8 +48,8 @@\n                 (\n                     cls.default_version,\n                     plat,\n-                    \"4d45336511484100ae4e2bab24542a8b86b12c8cb89230463593c60d08c4b8d3\",\n-                    \"3814407\",\n+                    \"bfc19b16e0f298742dd933289bd8057dd503f9ad0678310412d382800d48b3ae\",\n+                    \"3840814\",\n                 )\n             )\n             for plat in [\"macos_arm64\", \"macos_x86_64\", \"linux_x86_64\", \"linux_arm64\"]\n", "issue": "Pants `export` doesn't work on all machines\n**Describe the bug**\r\nI am able to run `./pants export ::` successfully on my machine. When I ran this through our CI system, I got a permission denied error when trying to clone a VCS requirement from our requirements.txt file. I know the CI worker had access to the repos because it cloned a repo to get this far (the VCS requirement is another internal repo).\r\n\r\nI already have the following added to `pants.toml`.\r\n```\r\n[subprocess-environment]\r\nenv_vars.add = [\"SSH_AUTH_SOCK\"]\r\n```\r\n\r\nI'm curious if it's possible this isn't getting passed through on my CI system and causing the export step to fail.\r\n\r\n**Pants version**\r\n2.11.0\r\n\r\n**OS**\r\nubuntu 20.04\r\n\r\n\n", "before_files": [{"content": "# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).\n# Licensed under the Apache License, Version 2.0 (see LICENSE).\n\nfrom __future__ import annotations\n\nimport dataclasses\nfrom dataclasses import dataclass\nfrom typing import Iterable, List, Mapping, Optional, Tuple\n\nfrom pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem\nfrom pants.backend.python.util_rules import pex_environment\nfrom pants.backend.python.util_rules.pex_environment import (\n    PexEnvironment,\n    PexSubsystem,\n    PythonExecutable,\n)\nfrom pants.core.util_rules import external_tool\nfrom pants.core.util_rules.external_tool import (\n    DownloadedExternalTool,\n    ExternalToolRequest,\n    TemplatedExternalTool,\n)\nfrom pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests\nfrom pants.engine.internals.selectors import MultiGet\nfrom pants.engine.platform import Platform\nfrom pants.engine.process import Process, ProcessCacheScope\nfrom pants.engine.rules import Get, collect_rules, rule\nfrom pants.option.global_options import GlobalOptions, ca_certs_path_to_file_content\nfrom pants.util.frozendict import FrozenDict\nfrom pants.util.logging import LogLevel\nfrom pants.util.meta import classproperty, frozen_after_init\nfrom pants.util.strutil import create_path_env_var\n\n\nclass PexCli(TemplatedExternalTool):\n    options_scope = \"pex-cli\"\n    name = \"pex\"\n    help = \"The PEX (Python EXecutable) tool (https://github.com/pantsbuild/pex).\"\n\n    default_version = \"v2.1.103\"\n    default_url_template = \"https://github.com/pantsbuild/pex/releases/download/{version}/pex\"\n    version_constraints = \">=2.1.103,<3.0\"\n\n    @classproperty\n    def default_known_versions(cls):\n        return [\n            \"|\".join(\n                (\n                    cls.default_version,\n                    plat,\n                    \"4d45336511484100ae4e2bab24542a8b86b12c8cb89230463593c60d08c4b8d3\",\n                    \"3814407\",\n                )\n            )\n            for plat in [\"macos_arm64\", \"macos_x86_64\", \"linux_x86_64\", \"linux_arm64\"]\n        ]\n\n\n@frozen_after_init\n@dataclass(unsafe_hash=True)\nclass PexCliProcess:\n    subcommand: tuple[str, ...]\n    extra_args: tuple[str, ...]\n    set_resolve_args: bool\n    description: str = dataclasses.field(compare=False)\n    additional_input_digest: Optional[Digest]\n    extra_env: Optional[FrozenDict[str, str]]\n    output_files: Optional[Tuple[str, ...]]\n    output_directories: Optional[Tuple[str, ...]]\n    python: Optional[PythonExecutable]\n    level: LogLevel\n    concurrency_available: int\n    cache_scope: ProcessCacheScope\n\n    def __init__(\n        self,\n        *,\n        subcommand: Iterable[str],\n        extra_args: Iterable[str],\n        description: str,\n        set_resolve_args: bool = True,\n        additional_input_digest: Optional[Digest] = None,\n        extra_env: Optional[Mapping[str, str]] = None,\n        output_files: Optional[Iterable[str]] = None,\n        output_directories: Optional[Iterable[str]] = None,\n        python: Optional[PythonExecutable] = None,\n        level: LogLevel = LogLevel.INFO,\n        concurrency_available: int = 0,\n        cache_scope: ProcessCacheScope = ProcessCacheScope.SUCCESSFUL,\n    ) -> None:\n        self.subcommand = tuple(subcommand)\n        self.extra_args = tuple(extra_args)\n        self.set_resolve_args = set_resolve_args\n        self.description = description\n        self.additional_input_digest = additional_input_digest\n        self.extra_env = FrozenDict(extra_env) if extra_env else None\n        self.output_files = tuple(output_files) if output_files else None\n        self.output_directories = tuple(output_directories) if output_directories else None\n        self.python = python\n        self.level = level\n        self.concurrency_available = concurrency_available\n        self.cache_scope = cache_scope\n        self.__post_init__()\n\n    def __post_init__(self) -> None:\n        if \"--pex-root-path\" in self.extra_args:\n            raise ValueError(\"`--pex-root` flag not allowed. We set its value for you.\")\n\n\nclass PexPEX(DownloadedExternalTool):\n    \"\"\"The Pex PEX binary.\"\"\"\n\n\n@rule\nasync def download_pex_pex(pex_cli: PexCli, platform: Platform) -> PexPEX:\n    pex_pex = await Get(DownloadedExternalTool, ExternalToolRequest, pex_cli.get_request(platform))\n    return PexPEX(digest=pex_pex.digest, exe=pex_pex.exe)\n\n\n@rule\nasync def setup_pex_cli_process(\n    request: PexCliProcess,\n    pex_pex: PexPEX,\n    pex_env: PexEnvironment,\n    python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,\n    global_options: GlobalOptions,\n    pex_subsystem: PexSubsystem,\n) -> Process:\n    tmpdir = \".tmp\"\n    gets: List[Get] = [Get(Digest, CreateDigest([Directory(tmpdir)]))]\n\n    cert_args = []\n    if global_options.ca_certs_path:\n        ca_certs_fc = ca_certs_path_to_file_content(global_options.ca_certs_path)\n        gets.append(Get(Digest, CreateDigest((ca_certs_fc,))))\n        cert_args = [\"--cert\", ca_certs_fc.path]\n\n    digests_to_merge = [pex_pex.digest]\n    digests_to_merge.extend(await MultiGet(gets))\n    if request.additional_input_digest:\n        digests_to_merge.append(request.additional_input_digest)\n    input_digest = await Get(Digest, MergeDigests(digests_to_merge))\n\n    global_args = [\n        # Ensure Pex and its subprocesses create temporary files in the the process execution\n        # sandbox. It may make sense to do this generally for Processes, but in the short term we\n        # have known use cases where /tmp is too small to hold large wheel downloads Pex is asked to\n        # perform. Making the TMPDIR local to the sandbox allows control via\n        # --local-execution-root-dir for the local case and should work well with remote cases where\n        # a remoting implementation has to allow for processes producing large binaries in a\n        # sandbox to support reasonable workloads. Communicating TMPDIR via --tmpdir instead of via\n        # environment variable allows Pex to absolutize the path ensuring subprocesses that change\n        # CWD can find the TMPDIR.\n        \"--tmpdir\",\n        tmpdir,\n    ]\n\n    if request.concurrency_available > 0:\n        global_args.extend([\"--jobs\", \"{pants_concurrency}\"])\n\n    verbosity_args = [f\"-{'v' * pex_subsystem.verbosity}\"] if pex_subsystem.verbosity > 0 else []\n\n    resolve_args = (\n        [*cert_args, \"--python-path\", create_path_env_var(pex_env.interpreter_search_paths)]\n        if request.set_resolve_args\n        else []\n    )\n    args = [\n        *request.subcommand,\n        *global_args,\n        *verbosity_args,\n        *resolve_args,\n        # NB: This comes at the end because it may use `--` passthrough args, # which must come at\n        # the end.\n        *request.extra_args,\n    ]\n\n    complete_pex_env = pex_env.in_sandbox(working_directory=None)\n    normalized_argv = complete_pex_env.create_argv(pex_pex.exe, *args, python=request.python)\n    env = {\n        **complete_pex_env.environment_dict(python_configured=request.python is not None),\n        **python_native_code.subprocess_env_vars,\n        **(request.extra_env or {}),\n        # If a subcommand is used, we need to use the `pex3` console script.\n        **({\"PEX_SCRIPT\": \"pex3\"} if request.subcommand else {}),\n    }\n\n    return Process(\n        normalized_argv,\n        description=request.description,\n        input_digest=input_digest,\n        env=env,\n        output_files=request.output_files,\n        output_directories=request.output_directories,\n        append_only_caches=complete_pex_env.append_only_caches,\n        level=request.level,\n        concurrency_available=request.concurrency_available,\n        cache_scope=request.cache_scope,\n    )\n\n\ndef rules():\n    return [\n        *collect_rules(),\n        *external_tool.rules(),\n        *pex_environment.rules(),\n    ]\n", "path": "src/python/pants/backend/python/util_rules/pex_cli.py"}], "after_files": [{"content": "# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).\n# Licensed under the Apache License, Version 2.0 (see LICENSE).\n\nfrom __future__ import annotations\n\nimport dataclasses\nfrom dataclasses import dataclass\nfrom typing import Iterable, List, Mapping, Optional, Tuple\n\nfrom pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem\nfrom pants.backend.python.util_rules import pex_environment\nfrom pants.backend.python.util_rules.pex_environment import (\n    PexEnvironment,\n    PexSubsystem,\n    PythonExecutable,\n)\nfrom pants.core.util_rules import external_tool\nfrom pants.core.util_rules.external_tool import (\n    DownloadedExternalTool,\n    ExternalToolRequest,\n    TemplatedExternalTool,\n)\nfrom pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests\nfrom pants.engine.internals.selectors import MultiGet\nfrom pants.engine.platform import Platform\nfrom pants.engine.process import Process, ProcessCacheScope\nfrom pants.engine.rules import Get, collect_rules, rule\nfrom pants.option.global_options import GlobalOptions, ca_certs_path_to_file_content\nfrom pants.util.frozendict import FrozenDict\nfrom pants.util.logging import LogLevel\nfrom pants.util.meta import classproperty, frozen_after_init\nfrom pants.util.strutil import create_path_env_var\n\n\nclass PexCli(TemplatedExternalTool):\n    options_scope = \"pex-cli\"\n    name = \"pex\"\n    help = \"The PEX (Python EXecutable) tool (https://github.com/pantsbuild/pex).\"\n\n    default_version = \"v2.1.107\"\n    default_url_template = \"https://github.com/pantsbuild/pex/releases/download/{version}/pex\"\n    version_constraints = \">=2.1.103,<3.0\"\n\n    @classproperty\n    def default_known_versions(cls):\n        return [\n            \"|\".join(\n                (\n                    cls.default_version,\n                    plat,\n                    \"bfc19b16e0f298742dd933289bd8057dd503f9ad0678310412d382800d48b3ae\",\n                    \"3840814\",\n                )\n            )\n            for plat in [\"macos_arm64\", \"macos_x86_64\", \"linux_x86_64\", \"linux_arm64\"]\n        ]\n\n\n@frozen_after_init\n@dataclass(unsafe_hash=True)\nclass PexCliProcess:\n    subcommand: tuple[str, ...]\n    extra_args: tuple[str, ...]\n    set_resolve_args: bool\n    description: str = dataclasses.field(compare=False)\n    additional_input_digest: Optional[Digest]\n    extra_env: Optional[FrozenDict[str, str]]\n    output_files: Optional[Tuple[str, ...]]\n    output_directories: Optional[Tuple[str, ...]]\n    python: Optional[PythonExecutable]\n    level: LogLevel\n    concurrency_available: int\n    cache_scope: ProcessCacheScope\n\n    def __init__(\n        self,\n        *,\n        subcommand: Iterable[str],\n        extra_args: Iterable[str],\n        description: str,\n        set_resolve_args: bool = True,\n        additional_input_digest: Optional[Digest] = None,\n        extra_env: Optional[Mapping[str, str]] = None,\n        output_files: Optional[Iterable[str]] = None,\n        output_directories: Optional[Iterable[str]] = None,\n        python: Optional[PythonExecutable] = None,\n        level: LogLevel = LogLevel.INFO,\n        concurrency_available: int = 0,\n        cache_scope: ProcessCacheScope = ProcessCacheScope.SUCCESSFUL,\n    ) -> None:\n        self.subcommand = tuple(subcommand)\n        self.extra_args = tuple(extra_args)\n        self.set_resolve_args = set_resolve_args\n        self.description = description\n        self.additional_input_digest = additional_input_digest\n        self.extra_env = FrozenDict(extra_env) if extra_env else None\n        self.output_files = tuple(output_files) if output_files else None\n        self.output_directories = tuple(output_directories) if output_directories else None\n        self.python = python\n        self.level = level\n        self.concurrency_available = concurrency_available\n        self.cache_scope = cache_scope\n        self.__post_init__()\n\n    def __post_init__(self) -> None:\n        if \"--pex-root-path\" in self.extra_args:\n            raise ValueError(\"`--pex-root` flag not allowed. We set its value for you.\")\n\n\nclass PexPEX(DownloadedExternalTool):\n    \"\"\"The Pex PEX binary.\"\"\"\n\n\n@rule\nasync def download_pex_pex(pex_cli: PexCli, platform: Platform) -> PexPEX:\n    pex_pex = await Get(DownloadedExternalTool, ExternalToolRequest, pex_cli.get_request(platform))\n    return PexPEX(digest=pex_pex.digest, exe=pex_pex.exe)\n\n\n@rule\nasync def setup_pex_cli_process(\n    request: PexCliProcess,\n    pex_pex: PexPEX,\n    pex_env: PexEnvironment,\n    python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,\n    global_options: GlobalOptions,\n    pex_subsystem: PexSubsystem,\n) -> Process:\n    tmpdir = \".tmp\"\n    gets: List[Get] = [Get(Digest, CreateDigest([Directory(tmpdir)]))]\n\n    cert_args = []\n    if global_options.ca_certs_path:\n        ca_certs_fc = ca_certs_path_to_file_content(global_options.ca_certs_path)\n        gets.append(Get(Digest, CreateDigest((ca_certs_fc,))))\n        cert_args = [\"--cert\", ca_certs_fc.path]\n\n    digests_to_merge = [pex_pex.digest]\n    digests_to_merge.extend(await MultiGet(gets))\n    if request.additional_input_digest:\n        digests_to_merge.append(request.additional_input_digest)\n    input_digest = await Get(Digest, MergeDigests(digests_to_merge))\n\n    global_args = [\n        # Ensure Pex and its subprocesses create temporary files in the the process execution\n        # sandbox. It may make sense to do this generally for Processes, but in the short term we\n        # have known use cases where /tmp is too small to hold large wheel downloads Pex is asked to\n        # perform. Making the TMPDIR local to the sandbox allows control via\n        # --local-execution-root-dir for the local case and should work well with remote cases where\n        # a remoting implementation has to allow for processes producing large binaries in a\n        # sandbox to support reasonable workloads. Communicating TMPDIR via --tmpdir instead of via\n        # environment variable allows Pex to absolutize the path ensuring subprocesses that change\n        # CWD can find the TMPDIR.\n        \"--tmpdir\",\n        tmpdir,\n    ]\n\n    if request.concurrency_available > 0:\n        global_args.extend([\"--jobs\", \"{pants_concurrency}\"])\n\n    verbosity_args = [f\"-{'v' * pex_subsystem.verbosity}\"] if pex_subsystem.verbosity > 0 else []\n\n    resolve_args = (\n        [*cert_args, \"--python-path\", create_path_env_var(pex_env.interpreter_search_paths)]\n        if request.set_resolve_args\n        else []\n    )\n    args = [\n        *request.subcommand,\n        *global_args,\n        *verbosity_args,\n        *resolve_args,\n        # NB: This comes at the end because it may use `--` passthrough args, # which must come at\n        # the end.\n        *request.extra_args,\n    ]\n\n    complete_pex_env = pex_env.in_sandbox(working_directory=None)\n    normalized_argv = complete_pex_env.create_argv(pex_pex.exe, *args, python=request.python)\n    env = {\n        **complete_pex_env.environment_dict(python_configured=request.python is not None),\n        **python_native_code.subprocess_env_vars,\n        **(request.extra_env or {}),\n        # If a subcommand is used, we need to use the `pex3` console script.\n        **({\"PEX_SCRIPT\": \"pex3\"} if request.subcommand else {}),\n    }\n\n    return Process(\n        normalized_argv,\n        description=request.description,\n        input_digest=input_digest,\n        env=env,\n        output_files=request.output_files,\n        output_directories=request.output_directories,\n        append_only_caches=complete_pex_env.append_only_caches,\n        level=request.level,\n        concurrency_available=request.concurrency_available,\n        cache_scope=request.cache_scope,\n    )\n\n\ndef rules():\n    return [\n        *collect_rules(),\n        *external_tool.rules(),\n        *pex_environment.rules(),\n    ]\n", "path": "src/python/pants/backend/python/util_rules/pex_cli.py"}]}