Add --remote-auth-plugin (pantsbuild#11503)

### Problem

Currently, remote caching/execution users can set authentication headers two ways:

1. Hardcoding via `--remote-store-headers` and `--remote-execution-headers`.
2. Storing in a file and using `--remote-oauth-bearer-token-path` and having Pants auto-set the relevant header.

However, this does not work for more complex cases where the headers should be dynamically set, such as wanting to make a network request to get an auth token, or consuming Pants options to determine the behavior. While users could write scripts to get the token and set it via env var, that does not scale very well and we want to provide a more robust first-class solution to dynamically setting auth headers.

### Solution

Add a simple plugin hook that allows users to write a function that does any arbitrary logic necessary and returns a standardized `AuthPluginResult` object. This will get called at the start of every Pants run.

Note that this plugin cannot use the Rules API, as it is used to determine the options that are used to initialize the Rust CommandRunner and Store. (Chicken and egg problem). Given this, the plugin runs pre-memoization so we do not need to worry about caching semantics here.

One expected user of this feature (Toolchain) needs the ability to consume arbitrary subsystems, so we rewire some things to pass an `Options` object to the `ExecutionOptions` constructor.

src/python/pants/bin/daemon_pants_runner.py

@@ -148,8 +148,7 @@ def single_daemonized_run(
         options_bootstrapper = OptionsBootstrapper.create(
             env=os.environ, args=sys.argv, allow_pantsrc=True
         )
-        bootstrap_options = options_bootstrapper.bootstrap_options
-        global_bootstrap_options = bootstrap_options.for_global_scope()
+        global_bootstrap_options = options_bootstrapper.bootstrap_options.for_global_scope()
 
         # Run using the pre-warmed Session.
         with self._stderr_logging(global_bootstrap_options):

src/python/pants/bin/local_pants_runner.py

@@ -5,7 +5,7 @@
 
 import logging
 import os
-from dataclasses import dataclass, replace
+from dataclasses import dataclass
 from typing import Mapping, Optional, Tuple
 
 from pants.base.build_environment import get_buildroot
@@ -27,14 +27,12 @@
 from pants.engine.target import RegisteredTargetTypes
 from pants.engine.unions import UnionMembership
 from pants.goal.run_tracker import RunTracker
-from pants.help.flag_error_help_printer import FlagErrorHelpPrinter
 from pants.help.help_info_extracter import HelpInfoExtracter
 from pants.help.help_printer import HelpPrinter
 from pants.init.engine_initializer import EngineInitializer, GraphScheduler, GraphSession
-from pants.init.options_initializer import BuildConfigInitializer, OptionsInitializer
+from pants.init.options_initializer import OptionsInitializer
 from pants.init.specs_calculator import calculate_specs
 from pants.option.arg_splitter import HelpRequest
-from pants.option.errors import UnknownFlagsError
 from pants.option.options import Options
 from pants.option.options_bootstrapper import OptionsBootstrapper
 from pants.util.contextutil import maybe_profiled
@@ -75,20 +73,13 @@ def _init_graph_session(
     ) -> GraphSession:
         native = Native()
         native.set_panic_handler()
-        graph_scheduler_helper = scheduler or EngineInitializer.setup_graph(
-            options_bootstrapper, build_config
-        )
-
-        try:
-            global_scope = options.for_global_scope()
-        except UnknownFlagsError as err:
-            cls._handle_unknown_flags(err, options_bootstrapper)
-            raise
-
+        graph_scheduler_helper = scheduler or EngineInitializer.setup_graph(options, build_config)
+        with OptionsInitializer.handle_unknown_flags(options_bootstrapper, raise_=True):
+            global_options = options.for_global_scope()
         return graph_scheduler_helper.new_session(
             run_id,
-            dynamic_ui=global_scope.dynamic_ui,
-            use_colors=global_scope.get("colors", True),
+            dynamic_ui=global_options.dynamic_ui,
+            use_colors=global_options.get("colors", True),
             session_values=SessionValues(
                 {
                     OptionsBootstrapper: options_bootstrapper,
@@ -98,15 +89,6 @@ def _init_graph_session(
             cancellation_latch=cancellation_latch,
         )
 
-    @staticmethod
-    def _handle_unknown_flags(err: UnknownFlagsError, options_bootstrapper: OptionsBootstrapper):
-        build_config = BuildConfigInitializer.get(options_bootstrapper)
-        # We need an options instance in order to get "did you mean" suggestions, but we know
-        # there are bad flags in the args, so we generate options with no flags.
-        no_arg_bootstrapper = replace(options_bootstrapper, args=("dummy_first_arg",))
-        options = OptionsInitializer.create(no_arg_bootstrapper, build_config)
-        FlagErrorHelpPrinter(options).handle_unknown_flags(err)
-
     @classmethod
     def create(
         cls,
@@ -124,14 +106,9 @@ def create(
         :param options_bootstrapper: The OptionsBootstrapper instance to reuse.
         :param scheduler: If being called from the daemon, a warmed scheduler to use.
         """
-        global_bootstrap_options = options_bootstrapper.bootstrap_options.for_global_scope()
-
-        build_config = BuildConfigInitializer.get(options_bootstrapper)
-        try:
-            options = OptionsInitializer.create(options_bootstrapper, build_config)
-        except UnknownFlagsError as err:
-            cls._handle_unknown_flags(err, options_bootstrapper)
-            raise
+        build_config, options = OptionsInitializer.create_with_build_config(
+            options_bootstrapper, raise_=True
+        )
 
         run_tracker = RunTracker(options)
         union_membership = UnionMembership.from_rules(build_config.union_rules)
@@ -147,16 +124,14 @@ def create(
             cancellation_latch,
         )
 
-        # Option values are usually computed lazily on demand,
-        # but command line options are eagerly computed for validation.
-        for scope in options.scope_to_flags.keys():
-            try:
+        # Option values are usually computed lazily on demand, but command line options are
+        # eagerly computed for validation.
+        with OptionsInitializer.handle_unknown_flags(options_bootstrapper, raise_=True):
+            for scope in options.scope_to_flags.keys():
                 options.for_scope(scope)
-            except UnknownFlagsError as err:
-                cls._handle_unknown_flags(err, options_bootstrapper)
-                raise
 
         # Verify configs.
+        global_bootstrap_options = options_bootstrapper.bootstrap_options.for_global_scope()
         if global_bootstrap_options.verify_config:
             options.verify_configs(options_bootstrapper.config)
 

src/python/pants/init/engine_initializer.py

@@ -32,7 +32,7 @@
 from pants.init import specs_calculator
 from pants.init.options_initializer import OptionsInitializer
 from pants.option.global_options import DEFAULT_EXECUTION_OPTIONS, ExecutionOptions
-from pants.option.options_bootstrapper import OptionsBootstrapper
+from pants.option.options import Options
 from pants.option.subsystem import Subsystem
 from pants.util.ordered_set import FrozenOrderedSet
 from pants.vcs.changed import rules as changed_rules
@@ -164,20 +164,20 @@ def _make_goal_map_from_rules(rules):
 
     @staticmethod
     def setup_graph(
-        options_bootstrapper: OptionsBootstrapper,
+        options: Options,
         build_configuration: BuildConfiguration,
         executor: Optional[PyExecutor] = None,
     ) -> GraphScheduler:
         native = Native()
         build_root = get_buildroot()
-        bootstrap_options = options_bootstrapper.bootstrap_options.for_global_scope()
+        bootstrap_options = options.bootstrap_option_values()
+        assert bootstrap_options is not None
         executor = executor or PyExecutor(
             *OptionsInitializer.compute_executor_arguments(bootstrap_options)
         )
         return EngineInitializer.setup_graph_extended(
-            options_bootstrapper,
             build_configuration,
-            ExecutionOptions.from_bootstrap_options(bootstrap_options),
+            ExecutionOptions.from_options(options),
             native=native,
             executor=executor,
             pants_ignore_patterns=OptionsInitializer.compute_pants_ignore(
@@ -190,11 +190,11 @@ def setup_graph(
             ca_certs_path=bootstrap_options.ca_certs_path,
             build_root=build_root,
             include_trace_on_error=bootstrap_options.print_stacktrace,
+            native_engine_visualize_to=bootstrap_options.native_engine_visualize_to,
         )
 
     @staticmethod
     def setup_graph_extended(
-        options_bootstrapper: OptionsBootstrapper,
         build_configuration: BuildConfiguration,
         execution_options: ExecutionOptions,
         native: Native,
@@ -208,14 +208,14 @@ def setup_graph_extended(
         ca_certs_path: Optional[str] = None,
         build_root: Optional[str] = None,
         include_trace_on_error: bool = True,
+        native_engine_visualize_to: Optional[str] = None,
     ) -> GraphScheduler:
         build_root = build_root or get_buildroot()
 
         rules = build_configuration.rules
         union_membership = UnionMembership.from_rules(build_configuration.union_rules)
         registered_target_types = RegisteredTargetTypes.create(build_configuration.target_types)
 
-        bootstrap_options = options_bootstrapper.bootstrap_options.for_global_scope()
         execution_options = execution_options or DEFAULT_EXECUTION_OPTIONS
 
         @rule
@@ -293,7 +293,7 @@ def ensure_optional_absolute_path(v: Optional[str]) -> Optional[str]:
             executor=executor,
             execution_options=execution_options,
             include_trace_on_error=include_trace_on_error,
-            visualize_to_dir=bootstrap_options.native_engine_visualize_to,
+            visualize_to_dir=native_engine_visualize_to,
         )
 
         return GraphScheduler(scheduler, goal_map)

src/python/pants/init/options_initializer.py

@@ -1,18 +1,22 @@
 # Copyright 2016 Pants project contributors (see CONTRIBUTORS.md).
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
+import dataclasses
 import logging
 import os
 import sys
-from typing import Optional, Tuple
+from contextlib import contextmanager
+from typing import Iterator, Optional, Tuple
 
 import pkg_resources
 
 from pants.base.build_environment import pants_version
 from pants.base.exceptions import BuildConfigurationError
 from pants.build_graph.build_configuration import BuildConfiguration
+from pants.help.flag_error_help_printer import FlagErrorHelpPrinter
 from pants.init.extension_loader import load_backends_and_plugins
 from pants.init.plugin_resolver import PluginResolver
+from pants.option.errors import UnknownFlagsError
 from pants.option.global_options import GlobalOptions
 from pants.option.option_value_container import OptionValueContainer
 from pants.option.options import Options
@@ -159,22 +163,47 @@ def create(
         build_configuration: BuildConfiguration,
     ) -> Options:
         global_bootstrap_options = options_bootstrapper.get_bootstrap_options().for_global_scope()
-
         if global_bootstrap_options.pants_version != pants_version():
             raise BuildConfigurationError(
                 f"Version mismatch: Requested version was {global_bootstrap_options.pants_version}, "
                 f"our version is {pants_version()}."
             )
 
         # Parse and register options.
-
         known_scope_infos = [
             si
             for optionable in build_configuration.all_optionables
             for si in optionable.known_scope_infos()
         ]
         options = options_bootstrapper.get_full_options(known_scope_infos)
-
         GlobalOptions.validate_instance(options.for_global_scope())
-
         return options
+
+    @classmethod
+    def create_with_build_config(
+        cls, options_bootstrapper: OptionsBootstrapper, *, raise_: bool
+    ) -> Tuple[BuildConfiguration, Options]:
+        build_config = BuildConfigInitializer.get(options_bootstrapper)
+        with OptionsInitializer.handle_unknown_flags(options_bootstrapper, raise_=raise_):
+            options = OptionsInitializer.create(options_bootstrapper, build_config)
+        return build_config, options
+
+    @classmethod
+    @contextmanager
+    def handle_unknown_flags(
+        cls, options_bootstrapper: OptionsBootstrapper, *, raise_: bool
+    ) -> Iterator[None]:
+        """If there are any unknown flags, print "Did you mean?" and possibly error."""
+        try:
+            yield
+        except UnknownFlagsError as err:
+            build_config = BuildConfigInitializer.get(options_bootstrapper)
+            # We need an options instance in order to get "did you mean" suggestions, but we know
+            # there are bad flags in the args, so we generate options with no flags.
+            no_arg_bootstrapper = dataclasses.replace(
+                options_bootstrapper, args=("dummy_first_arg",)
+            )
+            options = cls.create(no_arg_bootstrapper, build_config)
+            FlagErrorHelpPrinter(options).handle_unknown_flags(err)
+            if raise_:
+                raise err