Feature/token store

conan/api/conan_api.py

@@ -21,7 +21,7 @@
 from conan.api.subapi.remove import RemoveAPI
 from conan.api.subapi.upload import UploadAPI
 from conan.errors import ConanException
-from conan.internal.api.remotes.localdb import LocalDB
+from conan.internal.api.remotes.localdb import create_token_store
 from conan.internal.cache.cache import PkgCache
 from conan.internal.cache.home_paths import HomePaths
 from conan.internal.conan_app import ConanFileHelpers, CmdWrapper
@@ -176,7 +176,7 @@ def settings_yml(self):
         def remote_manager(self):
             if self._remote_manager is None:
                 home_folder = self._conan_api.home_folder
-                localdb = LocalDB(home_folder)
+                localdb = create_token_store(home_folder)
                 requester = self._conan_api._api_helpers.requester  # noqa
                 auth_manager = ConanApiAuthManager(requester, self._conan_api.home_folder, localdb,
                                                    self.global_conf)

conan/api/subapi/remotes.py

@@ -10,7 +10,7 @@
 from conan.internal.rest.remote_credentials import RemoteCredentials
 from conan.internal.rest.rest_client_local_recipe_index import add_local_recipes_index_remote, \
     remove_local_recipes_index_remote
-from conan.internal.api.remotes.localdb import LocalDB
+from conan.internal.api.remotes.localdb import create_token_store
 from conan.errors import ConanException
 from conan.internal.util.files import save, load
 
@@ -148,7 +148,7 @@ def remove(self, pattern):
         removed = _filter(remotes, pattern, only_enabled=False)
         remotes = [r for r in remotes if r not in removed]
         _save(self._remotes_file, remotes)
-        localdb = LocalDB(self._home_folder)
+        localdb = create_token_store(self._home_folder)
         for remote in removed:
             remove_local_recipes_index_remote(self._home_folder, remote)
             localdb.clean(remote_url=remote.url)
@@ -211,7 +211,7 @@ def rename(self, remote_name: str, new_name: str):
 
     def user_info(self, remote: Remote):
         # TODO: Review
-        localdb = LocalDB(self._home_folder)
+        localdb = create_token_store(self._home_folder)
         user_info = {}
         user, token, _ = localdb.get_login(remote.url)
         user_info["name"] = remote.name
@@ -255,20 +255,20 @@ def user_logout(self, remote: Remote):
 
         :param remote: The :ref:`Remote <conan.api.model.Remote>` object to logout
         """
-        localdb = LocalDB(self._home_folder)
+        localdb = create_token_store(self._home_folder)
         # The localdb only stores url + username + token, not remote name, so use URL as key
         localdb.clean(remote_url=remote.url)
 
     def user_set(self, remote: Remote, username):
         # TODO: Review
-        localdb = LocalDB(self._home_folder)
+        localdb = create_token_store(self._home_folder)
         if username == "":
             username = None
         localdb.store(username, token=None, refresh_token=None, remote_url=remote.url)
 
     def user_auth(self, remote: Remote, with_user=False, force=False):
         # TODO: Review
-        localdb = LocalDB(self._home_folder)
+        localdb = create_token_store(self._home_folder)
         if with_user:
             user, token, _ = localdb.get_login(remote.url)
             if not user:

conan/internal/api/remotes/localdb.py

@@ -5,6 +5,9 @@
 
 from conan.api.output import ConanOutput
 from conan.errors import ConanException
+from conan.internal.cache.home_paths import HomePaths
+from conan.internal.errors import scoped_traceback
+from conan.internal.loader import load_python_file
 
 REMOTES_USER_TABLE = "users_remotes"
 LOCALDB = ".conan.db"
@@ -15,6 +18,58 @@
                           warn_tag="risk")
 
 
+def create_token_store(home_folder):
+    """Return the token store. Plugin-backed if auth_remote plugin defines the
+    token_get/token_store/token_clean trio, else the default LocalDB (SQLite).
+    """
+    plugin_path = HomePaths(home_folder).auth_remote_plugin_path
+    if os.path.exists(plugin_path):
+        mod, _ = load_python_file(plugin_path)
+        get_fn = getattr(mod, "auth_remote_token_get", None)
+        store_fn = getattr(mod, "auth_remote_token_store", None)
+        clean_fn = getattr(mod, "auth_remote_token_clean", None)
+        if any((get_fn, store_fn, clean_fn)):
+            if not all((get_fn, store_fn, clean_fn)):
+                raise ConanException(
+                    "auth_remote plugin must define all of auth_remote_token_get, "
+                    "auth_remote_token_store, auth_remote_token_clean, or none")
+            return PluginTokenStore(get_fn, store_fn, clean_fn)
+    return LocalDB(home_folder)
+
+
+class PluginTokenStore:
+    """Duck-typed replacement for LocalDB. Delegates to auth_remote plugin functions."""
+
+    def __init__(self, get_fn, store_fn, clean_fn):
+        self._get = get_fn
+        self._store = store_fn
+        self._clean = clean_fn
+
+    @staticmethod
+    def _wrap(fn, *args):
+        try:
+            return fn(*args)
+        except Exception as e:
+            msg = scoped_traceback("Error while processing 'auth_remote.py' plugin",
+                                   e, scope="/extensions/plugins")
+            raise ConanException(msg)
+
+    def get_login(self, remote_url):
+        result = self._wrap(self._get, remote_url)
+        if not result:
+            return None, None, None
+        return result
+
+    def get_username(self, remote_url):
+        return self.get_login(remote_url)[0]
+
+    def store(self, user, token, refresh_token, remote_url):
+        self._wrap(self._store, remote_url, user, token, refresh_token)
+
+    def clean(self, remote_url=None):
+        self._wrap(self._clean, remote_url)
+
+
 class LocalDB:
 
     def __init__(self, dbfolder):

test/integration/configuration/test_auth_remote_token_plugin.py

@@ -0,0 +1,104 @@
+import os
+import json
+import textwrap
+
+from conan.test.utils.tools import TestClient
+from conan.internal.api.remotes.localdb import LOCALDB
+
+
+class TestAuthRemoteTokenPlugin:
+    """Token storage functions defined in auth_remote.py fully replace LocalDB."""
+
+    plugin = textwrap.dedent("""\
+        import json, os
+        _STORE = os.path.join(os.path.dirname(__file__), "..", "..", "tokens.json")
+
+        def _load():
+            if not os.path.exists(_STORE):
+                return {}
+            with open(_STORE) as f:
+                return json.load(f)
+
+        def _save(data):
+            with open(_STORE, "w") as f:
+                json.dump(data, f)
+
+        def auth_remote_plugin(remote, user=None):
+            return None, None  # no direct credentials; falls through
+
+        def auth_remote_token_get(remote_url):
+            d = _load().get(remote_url)
+            if not d:
+                return None, None, None
+            return d["user"], d["token"], d["refresh"]
+
+        def auth_remote_token_store(remote_url, user, token, refresh):
+            data = _load()
+            data[remote_url] = {"user": user, "token": token, "refresh": refresh}
+            _save(data)
+
+        def auth_remote_token_clean(remote_url):
+            data = _load()
+            if remote_url is None:
+                data.clear()
+            else:
+                data.pop(remote_url, None)
+            _save(data)
+        """)
+
+    def test_token_stored_only_in_plugin(self):
+        c = TestClient(default_server_user=True)
+        c.save_home({"extensions/plugins/auth_remote.py": self.plugin})
+        c.run("remote login default -p password")
+
+        assert not os.path.exists(os.path.join(c.cache_folder, LOCALDB))
+        tokens_file = os.path.join(c.cache_folder, "tokens.json")
+        data = json.load(open(tokens_file))
+        assert len(data) == 1
+        entry = next(iter(data.values()))
+        assert entry["user"] == "admin"
+        assert entry["token"]
+
+    def test_logout_clears_plugin_store(self):
+        c = TestClient(default_server_user=True)
+        c.save_home({"extensions/plugins/auth_remote.py": self.plugin})
+        c.run("remote login default -p password")
+        c.run("remote logout default")
+        data = json.load(open(os.path.join(c.cache_folder, "tokens.json")))
+        assert data == {}
+
+    def test_round_trip_token_reused(self):
+        """What the plugin stores at login must be what the plugin returns on later reads.
+        Verified two ways: the second command retrieves the same JWT the first one wrote,
+        and the CLI reports the remote as authenticated with no LocalDB fallback in play.
+        """
+        c = TestClient(default_server_user=True)
+        c.save_home({"extensions/plugins/auth_remote.py": self.plugin})
+
+        c.run("remote login default -p password")
+        stored = json.load(open(os.path.join(c.cache_folder, "tokens.json")))
+        remote_url, entry = next(iter(stored.items()))
+        assert "token" in entry
+
+        c.run("remote list-users")
+        assert "Username: admin" in c.out
+        assert "authenticated: True" in c.out
+        assert not os.path.exists(os.path.join(c.cache_folder, LOCALDB))
+
+        # Tamper: swap the stored token for a known-bad one; subsequent auth must fail,
+        # proving the token used comes from the plugin store (not from anywhere else).
+        stored[remote_url]["token"] = "not-a-valid-token"
+        with open(os.path.join(c.cache_folder, "tokens.json"), "w") as f:
+            json.dump(stored, f)
+        c.run("list * -r=default")
+        assert "Remote 'default' needs authentication, obtaining credentials" in c.out
+
+    def test_partial_token_functions_rejected(self):
+        c = TestClient(default_server_user=True)
+        broken = textwrap.dedent("""\
+            def auth_remote_token_get(remote_url):
+                return None, None, None
+            """)
+        c.save_home({"extensions/plugins/auth_remote.py": broken})
+        c.run("remote login default -p password", assert_error=True)
+        assert "must define all of auth_remote_token_get" in c.out