Convert authenticators to pydantic models

Author: tpoliaw

CHANGELOG.md

@@ -19,6 +19,7 @@ Write the date in place of the "Unreleased" in the case a new version is release
   of Closure Table to track ancestors and descendands of the nodes.
 - Shorter string representation of chunks in `ArrayClient`.
 - Refactored internal Zarr version detection
+- Convert authenticators into pydantic models
 
 ### Fixed
 

tiled/_tests/test_access_control.py

@@ -1,11 +1,13 @@
 import json
+import secrets
+from typing import Any
 
 import numpy
 import pytest
 from starlette.status import HTTP_403_FORBIDDEN
+from typing_extensions import TypedDict
 
-from tiled.authenticators import DictionaryAuthenticator
-from tiled.server.protocols import UserSessionState
+from tiled.server.protocols import InternalAuthenticator, UserSessionState
 
 from ..access_policies import NO_ACCESS
 from ..adapters.array import ArrayAdapter
@@ -531,22 +533,23 @@ def test_service_principal_access(tmpdir, sqlite_or_postgres_uri):
         assert list(sp_client) == ["x"]
 
 
-class CustomAttributesAuthenticator(DictionaryAuthenticator):
+UserAttributes = TypedDict(
+    "UserAttributes", {"password": str, "attributes": dict[str, Any]}, total=False
+)
+
+
+class CustomAttributesAuthenticator(InternalAuthenticator):
     """An example authenticator that enriches the stored user information."""
 
-    def __init__(self, users: dict, confirmation_message: str = ""):
-        self._users = users
-        super().__init__(
-            {username: user["password"] for username, user in users.items()},
-            confirmation_message,
-        )
+    users: dict[str, UserAttributes] = {}
 
     async def authenticate(self, username, password):
-        state = await super().authenticate(username, password)
-        if isinstance(state, UserSessionState):
-            # enrich the auth state
-            state.state["attributes"] = self._users[username].get("attributes", {})
-        return state
+        if (attrs := self.users.get(username)) and (pw := attrs.get("password")):
+            if secrets.compare_digest(pw, password):
+                state = UserSessionState(
+                    username, {"attributes": attrs.get("attributes", {})}
+                )
+                return state
 
 
 class CustomAttributesAccessPolicy:

tiled/_tests/test_authenticators.py

@@ -25,22 +25,22 @@
 ])
 # fmt: on
 @pytest.mark.parametrize("use_tls,use_ssl", [(False, False)])
+@pytest.mark.skipif(not TILED_TEST_LDAP, reason="Requires an LDAP container and TILED_TEST_LDAP to be set")
 def test_LDAPAuthenticator_01(use_tls, use_ssl, ldap_server_address, ldap_server_port):
     """
     Basic test for ``LDAPAuthenticator``.
 
     TODO: The test could be extended with enabled TLS or SSL, but it requires configuration
     of the LDAP server.
     """
-    if not TILED_TEST_LDAP:
-        pytest.skip("Run an LDAP container and set TILED_TEST_LDAP to run")
-    authenticator = LDAPAuthenticator(
-        ldap_server_address,
-        ldap_server_port,
-        bind_dn_template="cn={username},ou=users,dc=example,dc=org",
-        use_tls=use_tls,
-        use_ssl=use_ssl,
-    )
+
+    params = dict(server_address=ldap_server_address,
+                  bind_dn_template="cn={username},ou=users,dc=example,dc=org",
+                  use_ssl=use_ssl,
+                  use_tls=use_tls)
+    if ldap_server_port is not None:
+        params["server_port"] = ldap_server_port
+    authenticator = LDAPAuthenticator(**params)
 
     async def testing():
         assert (await authenticator.authenticate("user01", "password1")).user_name == "user01"