Convert authenticators to pydantic models

Author: tpoliaw

CHANGELOG.md

@@ -27,6 +27,7 @@ tiled catalog upgrade-database [postgresql://.. | sqlite:///...]
 - Refactored internal Zarr version detection
 - For compatibility with older clients, do not require metadata updates to include
   an `access_blob` in the body of the request.
+- Convert authenticators into pydantic models
 
 ### Fixed
 

tiled/_tests/test_access_control.py

@@ -1,11 +1,13 @@
 import json
+import secrets
+from typing import Any, Optional
 
 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,
-        )
-
-    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
+    users: dict[str, UserAttributes] = {}
+
+    async def authenticate(self, username, password) -> Optional[UserSessionState]:
+        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,21 +25,21 @@
 ])
 # 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,
+        server_address=ldap_server_address,
         bind_dn_template="cn={username},ou=users,dc=example,dc=org",
-        use_tls=use_tls,
         use_ssl=use_ssl,
+        use_tls=use_tls,
+        server_port=ldap_server_port
     )
 
     async def testing():
@@ -49,3 +49,19 @@ async def testing():
         assert (await authenticator.authenticate("user02", "password2a")) is None
 
     asyncio.run(testing())
+
+
+def test_ldap_port_validation():
+    # given port can be none but will be replaced with a default
+    auth = LDAPAuthenticator(server_address="http://ldap.example.com", server_port=None)
+    assert auth.server_port is not None
+
+
+def test_auth_server_list_wrapping():
+    auth = LDAPAuthenticator(server_address="http://ldap.example.com", server_port=None)
+    assert auth.server_address_list == ["http://ldap.example.com"]
+
+
+def test_list_of_addresses_not_nested_into_extra_list():
+    auth = LDAPAuthenticator(server_address=["http://ldap.example.com"])
+    assert auth.server_address_list == ["http://ldap.example.com"]