Add auth_plugin with tests

conan/cli/commands/remote.py

@@ -196,14 +196,16 @@ def remote_login(conan_api, parser, subparser, *args):
         raise ConanException("There are no remotes matching the '{}' pattern".format(args.remote))
 
     creds = RemoteCredentials(conan_api.cache_folder, conan_api.config.global_conf)
-    user, password = creds.auth(args.remote, args.username, args.password)
-    if args.username is not None and args.username != user:
-        raise ConanException(f"User '{args.username}' doesn't match user '{user}' in "
-                             f"credentials.json or environment variables")
 
     ret = OrderedDict()
     for r in remotes:
         previous_info = conan_api.remotes.user_info(r)
+
+        user, password = creds.auth(r, args.username, args.password)
+        if args.username is not None and args.username != user:
+            raise ConanException(f"User '{args.username}' doesn't match user '{user}' in "
+                                 f"credentials.json or environment variables")
+
         conan_api.remotes.user_login(r, user, password)
         info = conan_api.remotes.user_info(r)
         ret[r.name] = {"previous_info": previous_info, "info": info}

conan/internal/cache/home_paths.py

@@ -55,6 +55,10 @@ def profiles_path(self):
     def profile_plugin_path(self):
         return os.path.join(self._home, _EXTENSIONS_FOLDER, _PLUGINS, "profile.py")
 
+    @property
+    def auth_plugin_path(self):
+        return os.path.join(self._home, _EXTENSIONS_FOLDER, _PLUGINS, "auth.py")
+
     @property
     def sign_plugin_path(self):
         return os.path.join(self._home, _EXTENSIONS_FOLDER, _PLUGINS, "sign", "sign.py")

conans/client/rest/remote_credentials.py

@@ -4,16 +4,19 @@
 
 from jinja2 import Template
 
+from conan.internal.cache.home_paths import HomePaths
+
+from conans.client.loader import load_python_file
 from conan.api.output import ConanOutput
 from conans.client.userio import UserInput
-from conans.errors import ConanException
+from conans.errors import ConanException, scoped_traceback
 from conans.util.files import load
 
-
 class RemoteCredentials:
     def __init__(self, cache_folder, global_conf):
         self._global_conf = global_conf
         self._urls = {}
+        self.auth_plugin_path = HomePaths(cache_folder).auth_plugin_path
         creds_path = os.path.join(cache_folder, "credentials.json")
         if not os.path.exists(creds_path):
             return
@@ -32,24 +35,36 @@ def auth(self, remote, user=None, password=None):
         if user is not None and password is not None:
             return user, password
 
-        # First prioritize the cache "credentials.json" file
-        creds = self._urls.get(remote)
+        # First get the auth_plugin
+        auth_plugin = _load_auth_plugin(self.auth_plugin_path)
+        if auth_plugin is not None:
+            try:
+                plugin_user, plugin_password = auth_plugin(remote, user=user, password=password)
+            except Exception as e:
+                msg = f"Error while processing 'auth.py' plugin"
+                msg = scoped_traceback(msg, e, scope="/extensions/plugins")
+                raise ConanException(msg)
+            if plugin_user and plugin_password:
+                return plugin_user, plugin_password
+
+        # Then prioritize the cache "credentials.json" file
+        creds = self._urls.get(remote.name)
         if creds is not None:
             try:
                 return creds["user"], creds["password"]
             except KeyError as e:
                 raise ConanException(f"Authentication error, wrong credentials.json: {e}")
 
         # Then, check environment definition
-        env_user, env_passwd = self._get_env(remote, user)
+        env_user, env_passwd = self._get_env(remote.name, user)
         if env_passwd is not None:
             if env_user is None:
                 raise ConanException("Found password in env-var, but not defined user")
             return env_user, env_passwd
 
         # If not found, then interactive prompt
         ui = UserInput(self._global_conf.get("core:non_interactive", check_type=bool))
-        input_user, input_password = ui.request_login(remote, user)
+        input_user, input_password = ui.request_login(remote.name, user)
         return input_user, input_password
 
     @staticmethod
@@ -66,3 +81,9 @@ def _get_env(remote, user):
         if passwd:
             ConanOutput().info("Got password '******' from environment")
         return user, passwd
+
+def _load_auth_plugin(auth_plugin_path):
+    if os.path.exists(auth_plugin_path):
+        mod, _ = load_python_file(auth_plugin_path)
+        if hasattr(mod, "auth_plugin"):
+            return mod.auth_plugin

test/integration/configuration/test_auth_plugin.py

@@ -0,0 +1,60 @@
+import os
+import textwrap
+
+import pytest
+
+from conan.test.utils.tools import TestClient
+from conans.util.files import save
+
+
+class TestErrorsAuthPlugin:
+    """ when the plugin fails, we want a clear message and a helpful trace
+    """
+    def test_error_profile_plugin(self):
+        c = TestClient(default_server_user=True)
+        auth_plugin = textwrap.dedent("""\
+            def auth_plugin(remote, user=None, password=None):
+                raise Exception("Test Error")
+            """)
+        save(os.path.join(c.cache.plugins_path, "auth.py"), auth_plugin)
+        c.run("remote logout default")
+        c.run("remote login default", assert_error=True)
+        assert "Error while processing 'auth.py' plugin" in c.out
+        assert "Test Error" in c.out
+
+    def test_remove_plugin_file(self):
+        c = TestClient()
+        c.run("version")  # to trigger the creation
+        os.remove(os.path.join(c.cache.plugins_path, "auth.py"))
+        c.run("remote add default http://fake")
+        c.run("remote login default", assert_error=True)
+        assert "ERROR: The 'auth.py' plugin file doesn't exist" in c.out
+
+    @pytest.mark.parametrize("password", ["password", "bad-password"])
+    def test_profile_plugin_direct_credentials(self, password):
+        should_fail = password == "bad-password"
+        c = TestClient(default_server_user=True)
+        auth_plugin = textwrap.dedent(f"""\
+            def auth_plugin(remote, user=None, password=None):
+                return "admin", "{password}"
+            """)
+        save(os.path.join(c.cache.plugins_path, "auth.py"), auth_plugin)
+        c.run("remote logout default")
+        c.run("remote login default", assert_error=should_fail)
+        if should_fail:
+            assert "ERROR: Wrong user or password. [Remote: default]" in c.out
+        else:
+            assert "Changed user of remote 'default' from 'None' (anonymous) to 'admin' (authenticated)" in c.out
+
+    def test_profile_plugin_fallback(self):
+        c = TestClient(default_server_user=True)
+        auth_plugin = textwrap.dedent("""\
+                def auth_plugin(remote, user=None, password=None):
+                    return None, None
+                """)
+        save(os.path.join(c.cache.plugins_path, "auth.py"), auth_plugin)
+        c.run("remote logout default")
+        c.run("remote login default")
+        # As the auth plugin is not returning any password the code is falling back to the rest of
+        # the input methods in this case the stdin provided by TestClient.
+        assert "Changed user of remote 'default' from 'None' (anonymous) to 'admin' (authenticated)" in c.out