Feature/credentials json

conan/cli/commands/remote.py

@@ -7,7 +7,8 @@
 from conan.cli.command import conan_command, conan_subcommand, OnceArgument
 from conan.cli.commands.list import remote_color, error_color, recipe_color, \
     reference_color
-from conans.client.userio import UserInput
+from conans.client.cache.cache import ClientCache
+from conans.client.rest.remote_credentials import RemoteCredentials
 from conan.errors import ConanException
 
 
@@ -166,8 +167,8 @@ def remote_login(conan_api, parser, subparser, *args):
     """
     subparser.add_argument("remote", help="Pattern or name of the remote to login into. "
                                           "The pattern uses 'fnmatch' style wildcards.")
-    subparser.add_argument("username", help='Username')
-    subparser.add_argument("-p", "--password", nargs='?', const="", type=str, action=OnceArgument,
+    subparser.add_argument("username", nargs="?", help='Username')
+    subparser.add_argument("-p", "--password", nargs='?', type=str, action=OnceArgument,
                            help='User password. Use double quotes if password with spacing, '
                                 'and escape quotes if existing. If empty, the password is '
                                 'requested interactively (not exposed)')
@@ -177,15 +178,17 @@ def remote_login(conan_api, parser, subparser, *args):
     if not remotes:
         raise ConanException("There are no remotes matching the '{}' pattern".format(args.remote))
 
-    password = args.password
-    if not password:
-        ui = UserInput(conan_api.config.get("core:non_interactive"))
-        _, password = ui.request_login(remote_name=args.remote, username=args.username)
+    cache = ClientCache(conan_api.cache_folder)
+    creds = RemoteCredentials(cache)
+    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)
-        conan_api.remotes.login(r, args.username, password)
+        conan_api.remotes.login(r, user, password)
         info = conan_api.remotes.user_info(r)
         ret[r.name] = {"previous_info": previous_info, "info": info}
 

conans/client/rest/auth_manager.py

@@ -16,7 +16,7 @@
 
 from conan.api.output import ConanOutput
 from conans.client.cmd.user import update_localdb
-from conans.client.userio import UserInput
+from conans.client.rest.remote_credentials import RemoteCredentials
 from conans.errors import AuthenticationException, ConanException, ForbiddenException
 
 LOGIN_RETRIES = 3
@@ -70,8 +70,8 @@ def _retry_with_new_token(self, user, remote, method_name, *args, **kwargs):
         we can get a valid token from api_client. If a token is returned,
         credentials are stored in localdb and rest method is called"""
         for _ in range(LOGIN_RETRIES):
-            ui = UserInput(self._cache.new_config.get("core:non_interactive", check_type=bool))
-            input_user, input_password = ui.request_login(remote.name, user)
+            creds = RemoteCredentials(self._cache)
+            input_user, input_password = creds.auth(remote.name)
             try:
                 self._authenticate(remote, input_user, input_password)
             except AuthenticationException:

conans/client/rest/remote_credentials.py

@@ -0,0 +1,65 @@
+import json
+import os
+import platform
+
+from jinja2 import Template
+
+from conan.api.output import ConanOutput
+from conans.client.userio import UserInput
+from conans.errors import ConanException
+from conans.util.files import load
+
+
+class RemoteCredentials:
+    def __init__(self, cache):
+        self._cache = cache
+        self._urls = {}
+        creds_path = os.path.join(cache.cache_folder, "credentials.json")
+        if not os.path.exists(creds_path):
+            return
+        template = Template(load(creds_path))
+        content = template.render({"platform": platform, "os": os})
+        content = json.loads(content)
+
+        self._urls = {credentials["remote"]: {"user": credentials["user"],
+                                              "password": credentials["password"]}
+                      for credentials in content["credentials"]}
+
+    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)
+        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)
+        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._cache.new_config.get("core:non_interactive", check_type=bool))
+        input_user, input_password = ui.request_login(remote, user)
+        return input_user, input_password
+
+    @staticmethod
+    def _get_env(remote, user):
+        """
+        Try get creds from env-vars
+        """
+        remote = remote.replace("-", "_").upper()
+        if user is None:
+            user = os.getenv(f"CONAN_LOGIN_USERNAME_{remote}") or os.getenv("CONAN_LOGIN_USERNAME")
+            if user:
+                ConanOutput().info("Got username '%s' from environment" % user)
+        passwd = os.getenv(f"CONAN_PASSWORD_{remote}") or os.getenv("CONAN_PASSWORD")
+        if passwd:
+            ConanOutput().info("Got password '******' from environment")
+        return user, passwd

conans/client/userio.py

@@ -70,22 +70,14 @@ def request_login(self, remote_name, username=None):
         """Request user to input their name and password
         :param remote_name:
         :param username If username is specified it only request password"""
-
+        self._raise_if_non_interactive()
         if not username:
-            if self._interactive:
-                self._out.write("Remote '%s' username: " % remote_name)
-            username = self._get_env_username(remote_name)
-            if not username:
-                self._raise_if_non_interactive()
-                username = self.get_username()
-
-        if self._interactive:
-            self._out.write('Please enter a password for "%s" account: ' % username)
+            self._out.write("Remote '%s' username: " % remote_name)
+            username = self.get_username()
+
+        self._out.write('Please enter a password for "%s" account: ' % username)
         try:
-            pwd = self._get_env_password(remote_name)
-            if not pwd:
-                self._raise_if_non_interactive()
-                pwd = self.get_password()
+            pwd = self.get_password()
         except ConanException:
             raise
         except Exception as e:
@@ -96,9 +88,9 @@ def get_username(self):
         """Overridable for testing purpose"""
         return self.raw_input()
 
-    def get_password(self):
+    @staticmethod
+    def get_password():
         """Overridable for testing purpose"""
-        self._raise_if_non_interactive()
         try:
             return getpass.getpass("")
         except BaseException:  # For KeyboardInterrupt too
@@ -139,26 +131,3 @@ def request_boolean(self, msg, default_option=None):
             else:
                 self._out.error("%s is not a valid answer" % s)
         return ret
-
-    def _get_env_password(self, remote_name):
-        """
-        Try CONAN_PASSWORD_REMOTE_NAME or CONAN_PASSWORD or return None
-        """
-        remote_name = remote_name.replace("-", "_").upper()
-        var_name = "CONAN_PASSWORD_%s" % remote_name
-        ret = os.getenv(var_name, None) or os.getenv("CONAN_PASSWORD", None)
-        if ret:
-            self._out.info("Got password '******' from environment")
-        return ret
-
-    def _get_env_username(self, remote_name):
-        """
-        Try CONAN_LOGIN_USERNAME_REMOTE_NAME or CONAN_LOGIN_USERNAME or return None
-        """
-        remote_name = remote_name.replace("-", "_").upper()
-        var_name = "CONAN_LOGIN_USERNAME_%s" % remote_name
-        ret = os.getenv(var_name, None) or os.getenv("CONAN_LOGIN_USERNAME", None)
-
-        if ret:
-            self._out.info("Got username '%s' from environment" % ret)
-        return ret

conans/test/integration/remote/test_remote_file_credentials.py

@@ -0,0 +1,44 @@
+import json
+import os
+
+import pytest
+
+from conans.test.assets.genconanfile import GenConanfile
+from conans.test.utils.tools import TestClient, TestServer
+from conans.util.files import save
+
+
+@pytest.fixture()
+def client():
+    test_server = TestServer()
+    c = TestClient(servers={"default": test_server})
+    c.save({"conanfile.py": GenConanfile("pkg", "0.1")})
+    c.run("create .")
+    c.run("upload * -r=default -c", assert_error=True)
+    return c
+
+
+def test_remote_file_credentials(client):
+    c = client
+    content = {"credentials": [{"remote": "default", "user": "admin", "password": "password"}]}
+    save(os.path.join(c.cache_folder, "credentials.json"), json.dumps(content))
+    c.run("upload * -r=default -c")
+    # it works without problems!
+    assert "Uploading recipe" in c.out
+
+
+def test_remote_file_credentials_remote_login(client):
+    c = client
+    content = {"credentials": [{"remote": "default", "user": "admin", "password": "password"}]}
+    save(os.path.join(c.cache_folder, "credentials.json"), json.dumps(content))
+    c.run("remote login default")
+    assert "Changed user of remote 'default' from 'None' (anonymous) " \
+           "to 'admin' (authenticated)" in c.out
+
+
+def test_remote_file_credentials_error(client):
+    c = client
+    content = {"credentials": [{"remote": "default", "user": "admin", "password": "wrong"}]}
+    save(os.path.join(c.cache_folder, "credentials.json"), json.dumps(content))
+    c.run("upload * -r=default -c", assert_error=True)
+    assert "ERROR: Wrong user or password" in c.out

conans/test/integration/remote/token_refresh_test.py

@@ -9,6 +9,7 @@
 from conans.model.conf import ConfDefinition
 from conans.model.recipe_ref import RecipeReference
 from conans.test.utils.mocks import LocalDBMock
+from conans.test.utils.test_files import temp_folder
 
 common_headers = {"X-Conan-Server-Capabilities": "oauth_token,revisions",
                   "Content-Type": "application/json"}
@@ -83,13 +84,14 @@ def setUp(self):
         cache = Mock()
         cache.localdb = self.localdb
         cache.new_config = config
+        cache.cache_folder = temp_folder()
         self.auth_manager = ConanApiAuthManager(self.rest_client_factory, cache)
         self.remote = Remote("myremote", "myurl", True, True)
         self.ref = RecipeReference.loads("lib/1.0@conan/stable#myreciperev")
 
     def test_auth_with_token(self):
         """Test that if the capability is there, then we use the new endpoint"""
-        with mock.patch("conans.client.rest.auth_manager.UserInput.request_login",
+        with mock.patch("conans.client.rest.remote_credentials.UserInput.request_login",
                         return_value=("myuser", "mypassword")):
 
             self.auth_manager.call_rest_api_method(self.remote, "get_recipe", self.ref, ".",
@@ -102,7 +104,7 @@ def test_refresh_with_token(self):
         """The mock will raise 401 for a token value "expired" so it will try to refresh
         and only if the refresh endpoint is called, the value will be "refreshed_access_token"
         """
-        with mock.patch("conans.client.rest.auth_manager.UserInput.request_login",
+        with mock.patch("conans.client.rest.remote_credentials.UserInput.request_login",
                         return_value=("myuser", "mypassword")):
             self.localdb.access_token = "expired"
             self.localdb.refresh_token = "refresh_token"