WIP extracting the authenticator model changes from #928

Author: tpoliaw

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,25 @@ 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
+        print(f"Authenticating: {username=}, {password=}, {self.users=}")
+        if (attrs := self.users.get(username)) and (pw := attrs.get("password")):
+            if secrets.compare_digest(pw, password):
+                state = UserSessionState(
+                    username, {"attributes": attrs.get("attributes", {})}
+                )
+                print(f"{state=}")
+                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"

tiled/authenticators.py

@@ -5,12 +5,12 @@
 import re
 import secrets
 from collections.abc import Iterable
-from typing import Any, Mapping, Optional, cast
+from typing import Annotated, Any, Mapping, Optional, TypeVar, Union, cast
 
 import httpx
 from fastapi import APIRouter, Request
 from jose import JWTError, jwt
-from pydantic import Secret
+from pydantic import BeforeValidator, ConfigDict, Field, Secret
 from starlette.responses import RedirectResponse
 
 from .server.protocols import (
@@ -32,9 +32,6 @@ class DummyAuthenticator(InternalAuthenticator):
 
     """
 
-    def __init__(self, confirmation_message: str = ""):
-        self.confirmation_message = confirmation_message
-
     async def authenticate(self, username: str, password: str) -> UserSessionState:
         return UserSessionState(username, {})
 
@@ -46,31 +43,12 @@ class DictionaryAuthenticator(InternalAuthenticator):
     Check passwords from a dictionary of usernames mapped to passwords.
     """
 
-    configuration_schema = """
-$schema": http://json-schema.org/draft-07/schema#
-type: object
-additionalProperties: false
-properties:
-  users_to_password:
-    type: object
-    description: |
-      Mapping usernames to password. Environment variable expansion should be
-      used to avoid placing passwords directly in configuration.
-  confirmation_message:
-    type: string
-    description: May be displayed by client after successful login.
-"""
-
-    def __init__(
-        self, users_to_passwords: Mapping[str, str], confirmation_message: str = ""
-    ):
-        self._users_to_passwords = users_to_passwords
-        self.confirmation_message = confirmation_message
+    users_to_passwords: Mapping[str, str]
 
     async def authenticate(
         self, username: str, password: str
     ) -> Optional[UserSessionState]:
-        true_password = self._users_to_passwords.get(username)
+        true_password = self.users_to_passwords.get(username)
         if not true_password:
             # Username is not valid.
             return
@@ -79,26 +57,13 @@ async def authenticate(
 
 
 class PAMAuthenticator(InternalAuthenticator):
-    configuration_schema = """
-$schema": http://json-schema.org/draft-07/schema#
-type: object
-additionalProperties: false
-properties:
-  service:
-    type: string
-    description: PAM service. Default is 'login'.
-  confirmation_message:
-    type: string
-    description: May be displayed by client after successful login.
-"""
-
-    def __init__(self, service: str = "login", confirmation_message: str = ""):
+    service: str = "login"
+
+    def model_post_init(self, __context: Any):
         if not modules_available("pamela"):
             raise ModuleNotFoundError(
                 "This PAMAuthenticator requires the module 'pamela' to be installed."
             )
-        self.service = service
-        self.confirmation_message = confirmation_message
         # TODO Try to open a PAM session.
 
     async def authenticate(
@@ -115,47 +80,17 @@ async def authenticate(
 
 
 class OIDCAuthenticator(ExternalAuthenticator):
-    configuration_schema = """
-$schema": http://json-schema.org/draft-07/schema#
-type: object
-additionalProperties: false
-properties:
-  audience:
-    type: string
-  client_id:
-    type: string
-  client_secret:
-    type: string
-  well_known_uri:
-    type: string
-  confirmation_message:
-    type: string
-"""
-
-    def __init__(
-        self,
-        audience: str,
-        client_id: str,
-        client_secret: str,
-        well_known_uri: str,
-        confirmation_message: str = "",
-    ):
-        self._audience = audience
-        self._client_id = client_id
-        self._client_secret = Secret(client_secret)
-        self._well_known_url = well_known_uri
-        self.confirmation_message = confirmation_message
+    audience: str
+    client_id: str
+    client_secret: Secret[str]
+    well_known_uri: str
 
     @functools.cached_property
     def _config_from_oidc_url(self) -> dict[str, Any]:
-        response: httpx.Response = httpx.get(self._well_known_url)
+        response: httpx.Response = httpx.get(self.well_known_url)
         response.raise_for_status()
         return response.json()
 
-    @functools.cached_property
-    def client_id(self) -> str:
-        return self._client_id
-
     @functools.cached_property
     def id_token_signing_alg_values_supported(self) -> list[str]:
         return cast(
@@ -190,8 +125,8 @@ async def authenticate(self, request: Request) -> Optional[UserSessionState]:
         response = await exchange_code(
             self.token_endpoint,
             code,
-            self._client_id,
-            self._client_secret.get_secret_value(),
+            self.client_id,
+            self.client_secret.get_secret_value(),
             redirect_uri,
         )
         response_body = response.json()
@@ -207,7 +142,7 @@ async def authenticate(self, request: Request) -> Optional[UserSessionState]:
                 token=id_token,
                 key=keys,
                 algorithms=self.id_token_signing_alg_values_supported,
-                audience=self._audience,
+                audience=self.audience,
                 access_token=access_token,
             )
         except JWTError:
@@ -338,6 +273,18 @@ async def prepare_saml_from_fastapi_request(request: Request) -> Mapping[str, st
     return rv
 
 
+T = TypeVar("T")
+
+
+def one_or_many(value: Union[T, list[T]]) -> list[T]:
+    if isinstance(value, str) or not isinstance(value, Iterable):
+        return [value]
+    return list(value)
+
+
+OneOrMany = BeforeValidator(one_or_many)
+
+
 class LDAPAuthenticator(InternalAuthenticator):
     """
     The authenticator code is based on https://github.com/jupyterhub/ldapauthenticator
@@ -396,7 +343,7 @@ class LDAPAuthenticator(InternalAuthenticator):
                 "uid={username},ou=people,dc=wikimedia,dc=org",
                 "uid={username},ou=Developers,dc=wikimedia,dc=org"
                 ]
-    allowed_groups: list or None
+    allowed_groups: list
         List of LDAP group DNs that users could be members of to be granted access.
 
         If a user is in any one of the listed groups, then that user is granted access.
@@ -493,7 +440,10 @@ class LDAPAuthenticator(InternalAuthenticator):
 
         from bluesky_httpserver.authenticators import LDAPAuthenticator
         authenticator = LDAPAuthenticator(
-            "localhost", 1389, bind_dn_template="cn={username},ou=users,dc=example,dc=org", use_tls=False
+            server_address="localhost",
+            server_port=1389,
+            bind_dn_template="cn={username},ou=users,dc=example,dc=org",
+            use_tls=False
         )
         await authenticator.authenticate("user01", "password1")
         await authenticator.authenticate("user02", "password2")
@@ -521,79 +471,35 @@ class LDAPAuthenticator(InternalAuthenticator):
                 id: user02
     """
 
-    def __init__(
-        self,
-        server_address,
-        server_port=None,
-        *,
-        use_ssl=False,
-        use_tls=True,
-        connect_timeout=5,
-        receive_timeout=60,
-        bind_dn_template=None,
-        allowed_groups=None,
-        valid_username_regex=r"^[a-z][.a-z0-9_-]*$",
-        lookup_dn=False,
-        user_search_base=None,
-        user_attribute=None,
-        lookup_dn_search_filter="({login_attr}={login})",
-        lookup_dn_search_user=None,
-        lookup_dn_search_password=None,
-        lookup_dn_user_dn_attribute=None,
-        escape_userdn=False,
-        search_filter="",
-        attributes=None,
-        auth_state_attributes=None,
-        use_lookup_dn_username=True,
-        confirmation_message="",
-    ):
-        self.use_ssl = use_ssl
-        self.use_tls = use_tls
-        self.connect_timeout = connect_timeout
-        self.receive_timeout = receive_timeout
-        self.bind_dn_template = bind_dn_template
-        self.allowed_groups = allowed_groups
-        self.valid_username_regex = valid_username_regex
-        self.lookup_dn = lookup_dn
-        self.user_search_base = user_search_base
-        self.user_attribute = user_attribute
-        self.lookup_dn_search_filter = lookup_dn_search_filter
-        self.lookup_dn_search_user = lookup_dn_search_user
-        self.lookup_dn_search_password = lookup_dn_search_password
-        self.lookup_dn_user_dn_attribute = lookup_dn_user_dn_attribute
-        self.escape_userdn = escape_userdn
-        self.search_filter = search_filter
-        self.attributes = attributes if attributes else []
-        self.auth_state_attributes = (
-            auth_state_attributes if auth_state_attributes else []
-        )
-        self.use_lookup_dn_username = use_lookup_dn_username
-
-        if isinstance(server_address, str):
-            server_address_list = [server_address]
-        elif isinstance(server_address, Iterable):
-            server_address_list = list(server_address)
-        else:
-            raise TypeError(
-                f"Unsupported type of `server_address` (list): server_address={server_address} "
-                f"type(server_address)={type(server_address)}"
-            )
-        if not server_address_list:
-            raise ValueError(
-                "No servers are specified: 'server_address' is an empty list"
-            )
-
-        self.server_address_list = server_address_list
-        self.server_port = (
-            server_port if server_port is not None else self._server_port_default()
-        )
-        self.confirmation_message = confirmation_message
-
-    def _server_port_default(self):
-        if self.use_ssl:
-            return 636  # default SSL port for LDAP
-        else:
-            return 389  # default plaintext port for LDAP
+    model_config = ConfigDict(use_attribute_docstrings=True)
+
+    server_address_list: Annotated[list[str], OneOrMany, Field(alias="server_address")]
+    server_port: Annotated[
+        int,
+        Field(
+            default_factory=lambda data: port
+            if (port := data["server_port"] is not None)
+            else (636 if data["use_ssl"] else 389)
+        ),
+    ]
+    use_ssl: bool = False
+    use_tls: bool = True
+    connect_timeout: float = 5.0
+    receive_timeout: float = 60.0
+    bind_dn_template: Annotated[list[str], OneOrMany] = []
+    allowed_groups: list[str] = []
+    lookup_dn: bool = False
+    user_search_base: Optional[str] = None
+    user_attribute: Optional[str] = None
+    lookup_dn_search_filter: Optional[str] = "({login_attr}={login})"
+    lookup_dn_search_user: Optional[str] = None
+    lookup_dn_search_password: Optional[str] = None
+    lookup_dn_user_dn_attribute: Optional[str] = None
+    escape_userdn: bool = False
+    search_filter: str = ""
+    attributes: list[str] = []
+    auth_state_attributes: list[str] = []
+    use_lookup_dn_username: bool = True
 
     async def resolve_username(self, username_supplied_by_user):
         import ldap3