Add patron "Basic Token" authentication feature switch. (PP-431) (#1383)

README.md

@@ -161,6 +161,17 @@ To let the application know which database to use, set the `SIMPLIFIED_PRODUCTIO
 export SIMPLIFIED_PRODUCTION_DATABASE="postgresql://palace:test@localhost:5432/circ"
 ```
 
+##### Patron `Basic Token` authentication
+
+Enables/disables patron "basic token" authentication through setting the designated environment variable to any
+(case-insensitive) value of "true"/"yes"/"on"/"1" or "false"/"no"/"off"/"0", respectively.
+If the value is the empty string or the variable is not present in the environment, it is disabled by default.
+- `SIMPLIFIED_ENABLE_BASIC_TOKEN_AUTH`
+
+```sh
+export SIMPLIFIED_ENABLE_BASIC_TOKEN_AUTH=true
+```
+
 ##### Firebase Cloud Messaging
 
 For Firebase Cloud Messaging (FCM) support (e.g., for notifications), `one` (and only one) of the following should be set:

api/authentication/basic_token.py

@@ -23,6 +23,8 @@ class BasicTokenAuthenticationProvider(AuthenticationProvider):
     It is a companion to the basic authentication, and has no meaning without it.
     """
 
+    FLOW_TYPE = "http://thepalaceproject.org/authtype/basic-token"
+
     def __init__(
         self,
         _db: Session,
@@ -105,7 +107,7 @@ def remote_patron_lookup(self, _db):
 
     @property
     def flow_type(self) -> str:
-        return "http://thepalaceproject.org/authtype/basic-token"
+        return self.FLOW_TYPE
 
     @classmethod
     def description(cls) -> str:

api/authenticator.py

@@ -401,7 +401,9 @@ def register_basic_auth_provider(
         ):
             raise CannotLoadConfiguration("Two basic auth providers configured")
         self.basic_auth_provider = provider
-        if self.library is not None:
+        # TODO: We can remove the configuration test once
+        #  basic token authentication is fully deployed.
+        if self.library is not None and Configuration.basic_token_auth_is_enabled():
             self.access_token_authentication_provider = (
                 BasicTokenAuthenticationProvider(
                     self._db, self.library, self.basic_auth_provider

core/config.py

@@ -11,7 +11,7 @@
 # from this module, alongside CannotLoadConfiguration.
 from core.exceptions import IntegrationException
 
-from .util import LanguageCodes
+from .util import LanguageCodes, ansible_boolean
 from .util.datetime_helpers import to_utc, utc_now
 
 
@@ -40,6 +40,10 @@ class Configuration(ConfigurationConstants):
     DATABASE_TEST_ENVIRONMENT_VARIABLE = "SIMPLIFIED_TEST_DATABASE"
     DATABASE_PRODUCTION_ENVIRONMENT_VARIABLE = "SIMPLIFIED_PRODUCTION_DATABASE"
 
+    # TODO: We can remove this variable once basic token authentication is fully deployed.
+    # Patron token authentication enabled switch.
+    BASIC_TOKEN_AUTH_ENABLED_ENVVAR = "SIMPLIFIED_ENABLE_BASIC_TOKEN_AUTH"
+
     # Environment variables for Firebase Cloud Messaging (FCM) service account key
     FCM_CREDENTIALS_FILE_ENVIRONMENT_VARIABLE = "SIMPLIFIED_FCM_CREDENTIALS_FILE"
     FCM_CREDENTIALS_JSON_ENVIRONMENT_VARIABLE = "SIMPLIFIED_FCM_CREDENTIALS_JSON"
@@ -207,6 +211,26 @@ def database_url(cls):
         logging.info("Connecting to database: %s" % url_obj.__to_string__())
         return url
 
+    # TODO: We can remove this method once basic token authentication is fully deployed.
+    @classmethod
+    def basic_token_auth_is_enabled(cls) -> bool:
+        """Is basic token authentication enabled?
+
+        Return False, if the variable is unset or is an empty string.
+        Raises CannotLoadConfiguration, if the setting is invalid.
+        :raise CannotLoadConfiguration: If the setting contains an unsupported value.
+        """
+        try:
+            return ansible_boolean(
+                os.environ.get(cls.BASIC_TOKEN_AUTH_ENABLED_ENVVAR),
+                label=cls.BASIC_TOKEN_AUTH_ENABLED_ENVVAR,
+                default=False,
+            )
+        except (TypeError, ValueError) as e:
+            raise CannotLoadConfiguration(
+                f"Invalid value for {cls.BASIC_TOKEN_AUTH_ENABLED_ENVVAR} environment variable."
+            ) from e
+
     @classmethod
     def fcm_credentials(cls) -> Dict[str, str]:
         """Returns a dictionary containing Firebase Cloud Messaging credentials.

core/util/__init__.py

@@ -582,6 +582,39 @@ def chunks(lst, chunk_size, start_index=0):
         yield lst[i : i + chunk_size]
 
 
+def ansible_boolean(
+    value: Optional[str | bool],
+    label: Optional[str] = None,
+    default: Optional[bool] = None,
+) -> bool:
+    """Map Ansible "truthy" and "falsy" values to a Python boolean.
+
+    :param value: The value from which to map.
+    :param label: Optional name or label associated with the value.
+    :param default: Default result if value is empty string or None.
+    """
+    _value_label = f"Value of '{label}'" if label else "Value"
+    if default is not None and not isinstance(default, bool):
+        raise TypeError("'default' must be a boolean, when specified.")
+    if isinstance(value, bool):
+        return value
+    if value is None or value == "":
+        if default is not None:
+            return default
+        raise ValueError(
+            f"{_value_label} must be non-null and non-empty if no default is specified."
+        )
+    if not isinstance(value, str):
+        raise TypeError(f"{_value_label} must be a string or boolean.")
+
+    if value.upper() in ("TRUE", "T", "ON", "YES", "Y", "1"):
+        return True
+    if value.upper() in ("FALSE", "F", "OFF", "NO", "N", "0"):
+        return False
+
+    raise ValueError(f"{_value_label} does not map to True or False.")
+
+
 class ValuesMeta(type):
     """Metaclass to allow operators on simple constants defining classes"""
 

tests/api/test_authenticator.py

@@ -31,6 +31,7 @@
     Keyboards,
     LibraryIdentifierRestriction,
 )
+from api.authentication.basic_token import BasicTokenAuthenticationProvider
 from api.authenticator import (
     Authenticator,
     BaseSAMLAuthenticationProvider,
@@ -924,12 +925,18 @@ def test_authenticated_patron_bearer(
             assert saml.authenticated_patron.call_count == 1
 
     def test_authenticated_patron_bearer_access_token(
-        self, db: DatabaseTransactionFixture, mock_basic: MockBasicFixture
+        self,
+        db: DatabaseTransactionFixture,
+        mock_basic: MockBasicFixture,
     ):
         basic = mock_basic()
-        authenticator = LibraryAuthenticator(
-            _db=db.session, library=db.default_library(), basic_auth_provider=basic
-        )
+        # TODO: We can remove this patch once basic token authentication is fully deployed.
+        with patch.object(
+            Configuration, "basic_token_auth_is_enabled", return_value=True
+        ):
+            authenticator = LibraryAuthenticator(
+                _db=db.session, library=db.default_library(), basic_auth_provider=basic
+            )
         patron = db.patron()
         token = AccessTokenProvider.generate_token(db.session, patron, "pass")
         auth = Authorization(auth_type="bearer", token=token)
@@ -951,44 +958,56 @@ def test_authenticated_patron_unsupported_mechanism(
         assert UNSUPPORTED_AUTHENTICATION_MECHANISM == problem
 
     def test_get_credential_from_header(
-        self, db: DatabaseTransactionFixture, mock_basic: MockBasicFixture
+        self,
+        db: DatabaseTransactionFixture,
+        mock_basic: MockBasicFixture,
     ):
+        def get_library_authenticator(
+            basic_auth_provider: BasicAuthenticationProvider | None,
+        ) -> LibraryAuthenticator:
+            # TODO: We can remove this patch once basic token authentication is fully deployed.
+            with patch.object(
+                Configuration, "basic_token_auth_is_enabled", return_value=True
+            ):
+                return LibraryAuthenticator(
+                    _db=db.session,
+                    library=db.default_library(),
+                    basic_auth_provider=basic_auth_provider,
+                )
+
         basic = mock_basic()
 
         # We can pull the password out of a Basic Auth credential
         # if a Basic Auth authentication provider is configured.
-        authenticator = LibraryAuthenticator(
-            _db=db.session,
-            library=db.default_library(),
-            basic_auth_provider=basic,
-        )
+        authenticator = get_library_authenticator(basic_auth_provider=basic)
         credential = Authorization(auth_type="basic", data=dict(password="foo"))
         assert "foo" == authenticator.get_credential_from_header(credential)
 
         # We can't pull the password out if no basic auth provider
-        authenticator = LibraryAuthenticator(
-            _db=db.session,
-            library=db.default_library(),
-            basic_auth_provider=None,
-        )
+        authenticator = get_library_authenticator(basic_auth_provider=None)
         assert authenticator.get_credential_from_header(credential) is None
 
-        authenticator = LibraryAuthenticator(
-            _db=db.session,
-            library=db.default_library(),
-            basic_auth_provider=basic,
-        )
+        authenticator = get_library_authenticator(basic_auth_provider=basic)
         patron = db.patron()
         token = AccessTokenProvider.generate_token(db.session, patron, "passworx")
         credential = Authorization(auth_type="bearer", token=token)
         assert authenticator.get_credential_from_header(credential) == "passworx"
 
+    @pytest.mark.parametrize(
+        "token_auth_enabled, auth_count",
+        [
+            [True, 2],
+            [False, 1],
+        ],
+    )
     def test_create_authentication_document(
         self,
         db: DatabaseTransactionFixture,
         mock_basic: MockBasicFixture,
         announcement_fixture: AnnouncementFixture,
         library_fixture: LibraryFixture,
+        token_auth_enabled: bool,
+        auth_count: int,
     ):
         class MockAuthenticator(LibraryAuthenticator):
             """Mock the _geographic_areas method."""
@@ -1003,11 +1022,16 @@ def _geographic_areas(cls, library):
         library_settings = library_fixture.settings(library)
         basic = mock_basic()
         library.name = "A Fabulous Library"
-        authenticator = MockAuthenticator(
-            _db=db.session,
-            library=library,
-            basic_auth_provider=basic,
-        )
+        # TODO: We can remove this patch once basic token authentication is fully deployed.
+        with patch.object(
+            Configuration, "basic_token_auth_is_enabled"
+        ) as token_auth_enabled_method:
+            token_auth_enabled_method.return_value = token_auth_enabled
+            authenticator = MockAuthenticator(
+                _db=db.session,
+                library=library,
+                basic_auth_provider=basic,
+            )
 
         def annotate_authentication_document(library, doc, url_for):
             doc["modified"] = "Kilroy was here"
@@ -1103,7 +1127,17 @@ def annotate_authentication_document(library, doc, url_for):
             # The main thing we need to test is that the
             # authentication sub-documents are assembled properly and
             # placed in the right position.
-            [token_doc, basic_doc] = doc["authentication"]
+            # TODO: token doc will be here only when correct environment variable set to true.
+            # If basic token auth is enabled, then there should be two authentication
+            # mechanisms and the first should be for token auth.
+            authenticators = doc["authentication"]
+            assert auth_count > 0
+            assert auth_count == len(authenticators)
+            # TODO: We can remove this `if` block/restructure once basic token authentication is fully deployed.
+            if token_auth_enabled:
+                token_doc = authenticators[0]
+                assert BasicTokenAuthenticationProvider.FLOW_TYPE == token_doc["type"]
+            basic_doc = authenticators[auth_count - 1]
 
             expect_basic = basic.authentication_flow_document(db.session)
             assert expect_basic == basic_doc

tests/api/test_config.py

@@ -1,6 +1,7 @@
 import json
 import os
 from collections import Counter
+from contextlib import nullcontext as does_not_raise
 from unittest.mock import patch
 
 import pytest
@@ -196,3 +197,40 @@ def test_fcm_credentials(self, notifications_files_fixture):
             match=r"Cannot parse value of FCM credential environment variable .* as JSON.",
         ):
             Configuration.fcm_credentials()
+
+    @pytest.mark.parametrize(
+        "env_var_value, expected_result, raises_exception",
+        [
+            ["true", True, False],
+            ["True", True, False],
+            [None, False, False],
+            ["", False, False],
+            ["false", False, False],
+            ["False", False, False],
+            ["3", None, True],
+            ["X", None, True],
+        ],
+    )
+    @patch.object(os, "environ", new=dict())
+    def test_basic_token_auth_is_enabled(
+        self, env_var_value, expected_result, raises_exception
+    ):
+        env_var = Configuration.BASIC_TOKEN_AUTH_ENABLED_ENVVAR
+
+        # Simulate an unset environment variable with the `None` value.
+        if env_var_value is None:
+            del os.environ[env_var]
+        else:
+            os.environ[env_var] = env_var_value
+
+        expected_exception = (
+            pytest.raises(
+                CannotLoadConfiguration,
+                match=f"Invalid value for {env_var} environment variable.",
+            )
+            if raises_exception
+            else does_not_raise()
+        )
+
+        with expected_exception:
+            assert expected_result == Configuration.basic_token_auth_is_enabled()