test coverage for legacy login

jupyter_server/auth/identity.py

@@ -175,7 +175,7 @@ def _token_default(self):
 
     need_token = Bool(True)
 
-    async def get_user(self, handler: JupyterHandler) -> User | None:
+    def get_user(self, handler: JupyterHandler) -> User | None | Awaitable[User | None]:
         """Get the authenticated user for a request
 
         Must return a :class:`.jupyter_server.auth.User`,
@@ -185,6 +185,12 @@ async def get_user(self, handler: JupyterHandler) -> User | None:
 
         _may_ be a coroutine
         """
+        return self._get_user(handler)
+
+    # not sure how to have optional-async type signature
+    # on base class with `async def` without splitting it into two methods
+
+    async def _get_user(self, handler: JupyterHandler) -> User | None:
         if getattr(handler, "_jupyter_current_user", None):
             # already authenticated
             return handler._jupyter_current_user
@@ -200,11 +206,11 @@ async def get_user(self, handler: JupyterHandler) -> User | None:
         # because token is always explicit
         user = token_user or cookie_user
 
-        if token_user:
+        if user is not None and token_user is not None:
             # if token-authenticated, persist user_id in cookie
             # if it hasn't already been stored there
             if user != cookie_user:
-                self.set_login_cookie(handler, cast(User, user))
+                self.set_login_cookie(handler, user)
             # Record that the current request has been authenticated with a token.
             # Used in is_token_authenticated above.
             handler._token_authenticated = True
@@ -522,7 +528,7 @@ def process_login_form(self, handler: JupyterHandler) -> User | None:
             if new_password and self.allow_password_change:
                 config_dir = handler.settings.get("config_dir", "")
                 config_file = os.path.join(config_dir, "jupyter_server_config.json")
-                set_password(new_password, config_file=config_file)
+                self.hashed_password = set_password(new_password, config_file=config_file)
                 self.log.info(_i18n(f"Wrote hashed password to {config_file}"))
 
         return user
@@ -552,6 +558,13 @@ class LegacyIdentityProvider(PasswordIdentityProvider):
     # settings must be passed for
     settings = Dict()
 
+    @default("settings")
+    def _default_settings(self):
+        return {
+            "token": self.token,
+            "password": self.hashed_password,
+        }
+
     @default("login_handler_class")
     def _default_login_handler_class(self):
         from .login import LegacyLoginHandler
@@ -562,12 +575,15 @@ def _default_login_handler_class(self):
     def auth_enabled(self):
         return self.login_available
 
-    def get_user(self, handler):
-        return _backward_compat_user(self.login_handler_class.get_user(handler))
+    def get_user(self, handler: JupyterHandler) -> User | None:
+        user = self.login_handler_class.get_user(handler)
+        if user is None:
+            return None
+        return _backward_compat_user(user)
 
     @property
     def login_available(self):
-        return self.login_handler_class.login_available(self.settings)
+        return self.login_handler_class.get_login_available(self.settings)
 
     def should_check_origin(self, handler: JupyterHandler) -> bool:
         return self.login_handler_class.should_check_origin(handler)

jupyter_server/auth/login.py

@@ -106,7 +106,9 @@ def post(self):
                 if new_password and self.settings.get("allow_password_change"):
                     config_dir = self.settings.get("config_dir", "")
                     config_file = os.path.join(config_dir, "jupyter_server_config.json")
-                    set_password(new_password, config_file=config_file)
+                    self.identity_provider.hashed_password = self.settings[
+                        "password"
+                    ] = set_password(new_password, config_file=config_file)
                     self.log.info("Wrote hashed password to %s" % config_file)
             else:
                 self.set_status(401)

jupyter_server/auth/security.py

@@ -169,4 +169,5 @@ def set_password(password=None, config_file=None):
     hashed_password = passwd(password)
 
     with persist_config(config_file) as config:
-        config.ServerApp.password = hashed_password
+        config.IdentityProvider.hashed_password = hashed_password
+    return hashed_password

jupyter_server/serverapp.py

@@ -1104,15 +1104,21 @@ def _default_min_open_files_limit(self):
     def _warn_deprecated_config(self, change, clsname, new_name=None):
         if new_name is None:
             new_name = change.name
-        if (
-            clsname not in self.config
-            or change.name not in self.config[clsname]
-            or self.config[clsname][change.name] != change.new
-        ):
+        if clsname not in self.config or new_name not in self.config[clsname]:
+            # Deprecated config used, new config not used.
+            # Use deprecated config, warn about new name.
             self.log.warning(
                 f"ServerApp.{change.name} config is deprecated in 2.0. Use {clsname}.{new_name}."
             )
-            self.config[clsname][change.name] = change.new
+            self.config[clsname][new_name] = change.new
+        else:
+            # Deprecated config used, new config also used.
+            # Warn only if the values differ.
+            # If the values are the same, assume intentional backward-compatible config.
+            if self.config[clsname][new_name] != change.new:
+                self.log.warning(
+                    f"Ignoring deprecated ServerApp.{change.name} config. Using {clsname}.{new_name}."
+                )
 
     @observe("password")
     def _deprecated_password(self, change):
@@ -1873,6 +1879,7 @@ def init_configurables(self):
         if self.identity_provider_class is LegacyIdentityProvider:
             # legacy config stored the password in tornado_settings
             self.tornado_settings["password"] = self.identity_provider.hashed_password
+            self.tornado_settings["token"] = self.identity_provider.token
 
         if self._token_set:
             self.log.warning(

tests/auth/test_legacy_login.py

@@ -0,0 +1,157 @@
+"""
+Test legacy login config via ServerApp.login_handler_class
+"""
+
+import json
+from functools import wraps
+from urllib.parse import urlencode
+
+import pytest
+from tornado.httpclient import HTTPClientError
+from tornado.httputil import parse_cookie, url_concat
+from traitlets.config import Config
+
+from jupyter_server.auth.identity import LegacyIdentityProvider
+from jupyter_server.auth.login import LoginHandler
+from jupyter_server.auth.security import passwd
+from jupyter_server.serverapp import ServerApp
+from jupyter_server.utils import url_path_join
+
+# Don't raise on deprecation warnings in this module testing deprecated behavior
+pytestmark = pytest.mark.filterwarnings("ignore::DeprecationWarning")
+
+
+def record_calls(f):
+    """Decorator to record call history"""
+    f._calls = []
+
+    @wraps(f)
+    def wrapped_f(*args, **kwargs):
+        f._calls.append((args, kwargs))
+        return f(*args, **kwargs)
+
+    return wrapped_f
+
+
+class CustomLoginHandler(LoginHandler):
+    @classmethod
+    @record_calls
+    def get_user(cls, handler):
+        header_user = handler.request.headers.get("test-user")
+        if header_user:
+            if header_user == "super":
+                return super().get_user(handler)
+            return header_user
+        else:
+            return None
+
+
+@pytest.fixture
+def jp_server_config():
+    cfg = Config()
+    cfg.ServerApp.login_handler_class = CustomLoginHandler
+    return cfg
+
+
+def test_legacy_identity_config(jp_serverapp):
+    # setting login_handler_class sets LegacyIdentityProvider
+    app = ServerApp()
+    idp = jp_serverapp.identity_provider
+    assert type(idp) is LegacyIdentityProvider
+    assert idp.login_available
+    assert idp.auth_enabled
+    assert idp.token
+    assert idp.get_handlers() == [
+        ("/login", idp.login_handler_class),
+        ("/logout", idp.logout_handler_class),
+    ]
+
+
+async def test_legacy_identity_api(jp_serverapp, jp_fetch):
+    response = await jp_fetch("/api/me", headers={"test-user": "pinecone"})
+    assert response.code == 200
+    model = json.loads(response.body.decode("utf8"))
+    assert model["identity"]["username"] == "pinecone"
+
+
+async def test_legacy_base_class(jp_serverapp, jp_fetch):
+    response = await jp_fetch("/api/me", headers={"test-user": "super"})
+    assert "Set-Cookie" in response.headers
+    cookie = response.headers["Set-Cookie"]
+    assert response.code == 200
+    model = json.loads(response.body.decode("utf8"))
+    user_id = model["identity"]["username"]  # a random uuid
+    assert user_id
+
+    response = await jp_fetch("/api/me", headers={"test-user": "super", "Cookie": cookie})
+    model2 = json.loads(response.body.decode("utf8"))
+    # second request, should trigger cookie auth
+    assert model2["identity"] == model["identity"]
+
+
+async def test_legacy_login(jp_serverapp, http_server_client, jp_base_url, jp_fetch):
+    login_url = url_path_join(jp_base_url, "login")
+    first = await http_server_client.fetch(login_url)
+    cookie_header = first.headers["Set-Cookie"]
+    xsrf = parse_cookie(cookie_header).get("_xsrf", "")
+    new_password = "super-secret"
+
+    async def login(form_fields):
+        form = {"_xsrf": xsrf}
+        form.update(form_fields)
+        try:
+            resp = await http_server_client.fetch(
+                login_url,
+                method="POST",
+                body=urlencode(form),
+                headers={"Cookie": cookie_header},
+                follow_redirects=False,
+            )
+        except HTTPClientError as e:
+            resp = e.response
+        assert resp.code == 302, "Should have returned a redirect!"
+        return resp
+
+    resp = await login(
+        dict(password=jp_serverapp.identity_provider.token, new_password=new_password)
+    )
+    cookie = resp.headers["Set-Cookie"]
+    id_resp = await jp_fetch("/api/me", headers={"test-user": "super", "Cookie": cookie})
+    assert id_resp.code == 200
+    model = json.loads(id_resp.body.decode("utf8"))
+    user_id = model["identity"]["username"]  # a random uuid
+
+    # verify password change with second login
+    resp2 = await login(dict(password=new_password))
+    cookie = resp.headers["Set-Cookie"]
+    id_resp = await jp_fetch("/api/me", headers={"test-user": "super", "Cookie": cookie})
+    assert id_resp.code == 200
+    model = json.loads(id_resp.body.decode("utf8"))
+    user_id2 = model["identity"]["username"]  # a random uuid
+    assert user_id2 == user_id
+
+
+def test_deprecated_config():
+    cfg = Config()
+    cfg.ServerApp.token = token = "asdf"
+    cfg.ServerApp.password = password = passwd("secrets")
+    app = ServerApp(config=cfg)
+    app.initialize([])
+    app.init_configurables()
+    assert app.identity_provider.token == token
+    assert app.token == token
+    assert app.identity_provider.hashed_password == password
+    assert app.password == password
+
+
+def test_deprecated_config_priority():
+    cfg = Config()
+    cfg.ServerApp.token = "ignored"
+    cfg.IdentityProvider.token = token = "idp_token"
+    cfg.ServerApp.password = passwd("ignored")
+    cfg.PasswordIdentityProvider.hashed_password = password = passwd("used")
+    app = ServerApp(config=cfg)
+    app.initialize([])
+    app.init_configurables()
+    assert app.identity_provider.token == token
+    assert app.identity_provider.hashed_password == password