Fix provider name handling in OIDC integration

pyproject.toml

@@ -93,7 +93,9 @@ ignore_errors = true
 
 [[tool.mypy.overrides]]
 module = [
+    "common.helpers",
     "tests.integration.*",
+    "tests.storage_service.test_helpers",
     "tests.storage_service.test_oidc",
 ]
 ignore_errors = false

storage_service/common/backends.py

@@ -41,10 +41,7 @@ def get_settings(self, attr, *args):
             if request:
                 provider_name = request.session.get("providername")
 
-                if (
-                    provider_name
-                    and provider_name in settings.OIDC_SECONDARY_PROVIDER_NAMES
-                ):
+                if provider_name and provider_name in settings.OIDC_PROVIDERS:
                     provider_settings = settings.OIDC_PROVIDERS.get(provider_name, {})
                     value = provider_settings.get(attr)
 

storage_service/common/helpers.py

@@ -0,0 +1,50 @@
+from os import environ
+from typing import Any
+from typing import Dict
+from typing import Iterable
+
+from django.core.exceptions import ImproperlyConfigured
+
+
+def get_env_variable(var_name: str) -> Any:
+    """Get the environment variable or return exception"""
+    try:
+        return environ[var_name]
+    except KeyError:
+        error_msg = "Set the %s environment variable" % var_name
+        raise ImproperlyConfigured(error_msg)
+
+
+def is_true(env_str: str) -> bool:
+    return env_str.lower() in ["true", "yes", "on", "1"]
+
+
+def get_oidc_secondary_providers(
+    oidc_secondary_provider_names: Iterable[str],
+) -> Dict[str, Dict[str, str]]:
+    providers = {}
+
+    for provider_name in oidc_secondary_provider_names:
+        provider_name = provider_name.strip().upper()
+        client_id = environ.get(f"OIDC_RP_CLIENT_ID_{provider_name}")
+        client_secret = environ.get(f"OIDC_RP_CLIENT_SECRET_{provider_name}")
+        authorization_endpoint = environ.get(
+            f"OIDC_OP_AUTHORIZATION_ENDPOINT_{provider_name}", ""
+        )
+        token_endpoint = environ.get(f"OIDC_OP_TOKEN_ENDPOINT_{provider_name}", "")
+        user_endpoint = environ.get(f"OIDC_OP_USER_ENDPOINT_{provider_name}", "")
+        jwks_endpoint = environ.get(f"OIDC_OP_JWKS_ENDPOINT_{provider_name}", "")
+        logout_endpoint = environ.get(f"OIDC_OP_LOGOUT_ENDPOINT_{provider_name}", "")
+
+        if client_id and client_secret:
+            providers[provider_name] = {
+                "OIDC_RP_CLIENT_ID": client_id,
+                "OIDC_RP_CLIENT_SECRET": client_secret,
+                "OIDC_OP_AUTHORIZATION_ENDPOINT": authorization_endpoint,
+                "OIDC_OP_TOKEN_ENDPOINT": token_endpoint,
+                "OIDC_OP_USER_ENDPOINT": user_endpoint,
+                "OIDC_OP_JWKS_ENDPOINT": jwks_endpoint,
+                "OIDC_OP_LOGOUT_ENDPOINT": logout_endpoint,
+            }
+
+    return providers

storage_service/storage_service/settings/base.py

@@ -9,11 +9,11 @@
 from typing import Dict
 from typing import List
 
+from common.helpers import get_oidc_secondary_providers
+from common.helpers import is_true
 from django.core.exceptions import ImproperlyConfigured
 from django.utils.translation import gettext_lazy as _
 
-from storage_service.settings.helpers import is_true
-
 from .components.s3 import *
 
 try:
@@ -568,45 +568,6 @@ def _get_settings_from_file(path):
 ######### OIDC CONFIGURATION #########
 OIDC_AUTHENTICATION = is_true(environ.get("SS_OIDC_AUTHENTICATION", ""))
 if OIDC_AUTHENTICATION:
-
-    def get_oidc_secondary_providers(oidc_secondary_provider_names):
-        providers = {}
-
-        for provider_name in oidc_secondary_provider_names:
-            provider_name = provider_name.strip()
-            client_id = environ.get(f"OIDC_RP_CLIENT_ID_{provider_name.upper()}")
-            client_secret = environ.get(
-                f"OIDC_RP_CLIENT_SECRET_{provider_name.upper()}"
-            )
-            authorization_endpoint = environ.get(
-                f"OIDC_OP_AUTHORIZATION_ENDPOINT_{provider_name.upper()}", ""
-            )
-            token_endpoint = environ.get(
-                f"OIDC_OP_TOKEN_ENDPOINT_{provider_name.upper()}", ""
-            )
-            user_endpoint = environ.get(
-                f"OIDC_OP_USER_ENDPOINT_{provider_name.upper()}", ""
-            )
-            jwks_endpoint = environ.get(
-                f"OIDC_OP_JWKS_ENDPOINT_{provider_name.upper()}", ""
-            )
-            logout_endpoint = environ.get(
-                f"OIDC_OP_LOGOUT_ENDPOINT_{provider_name.upper()}", ""
-            )
-
-            if client_id and client_secret:
-                providers[provider_name] = {
-                    "OIDC_RP_CLIENT_ID": client_id,
-                    "OIDC_RP_CLIENT_SECRET": client_secret,
-                    "OIDC_OP_AUTHORIZATION_ENDPOINT": authorization_endpoint,
-                    "OIDC_OP_TOKEN_ENDPOINT": token_endpoint,
-                    "OIDC_OP_USER_ENDPOINT": user_endpoint,
-                    "OIDC_OP_JWKS_ENDPOINT": jwks_endpoint,
-                    "OIDC_OP_LOGOUT_ENDPOINT": logout_endpoint,
-                }
-
-        return providers
-
     ALLOW_USER_EDITS = False
     INSTALLED_APPS += ["mozilla_django_oidc"]
 

storage_service/storage_service/settings/components/auth.py

@@ -2,7 +2,7 @@
 
 from os import environ
 
-from storage_service.settings.helpers import is_true
+from common.helpers import is_true
 
 PASSWORD_MINIMUM_LENGTH = 8
 try:

storage_service/storage_service/settings/local.py

@@ -3,8 +3,7 @@
 from os import environ
 
 import dj_database_url
-
-from storage_service.settings.helpers import get_env_variable
+from common.helpers import get_env_variable
 
 from .base import *
 

storage_service/storage_service/settings/production.py

@@ -3,9 +3,8 @@
 from os import environ
 
 import dj_database_url
-
-from storage_service.settings.helpers import get_env_variable
-from storage_service.settings.helpers import is_true
+from common.helpers import get_env_variable
+from common.helpers import is_true
 
 from .base import *
 

storage_service/storage_service/views.py

@@ -44,10 +44,7 @@ def get_settings(self, attr, *args):
             if request:
                 provider_name = request.session.get("providername")
 
-                if (
-                    provider_name
-                    and provider_name in settings.OIDC_SECONDARY_PROVIDER_NAMES
-                ):
+                if provider_name and provider_name in settings.OIDC_PROVIDERS:
                     provider_settings = settings.OIDC_PROVIDERS.get(provider_name, {})
                     value = provider_settings.get(attr)
 
@@ -108,7 +105,7 @@ def get_oidc_logout_url(request):
     if request:
         provider_name = request.session.get("providername")
 
-        if provider_name and provider_name in settings.OIDC_SECONDARY_PROVIDER_NAMES:
+        if provider_name and provider_name in settings.OIDC_PROVIDERS:
             provider_settings = settings.OIDC_PROVIDERS.get(provider_name, {})
             end_session_endpoint = provider_settings.get("OIDC_OP_LOGOUT_ENDPOINT")
 

tests/integration/docker-compose.yml

@@ -40,7 +40,7 @@ services:
       OIDC_OP_USER_ENDPOINT: "http://keycloak:8080/realms/demo/protocol/openid-connect/userinfo"
       OIDC_OP_JWKS_ENDPOINT: "http://keycloak:8080/realms/demo/protocol/openid-connect/certs"
       OIDC_OP_LOGOUT_ENDPOINT: "http://keycloak:8080/realms/demo/protocol/openid-connect/logout"
-      OIDC_SECONDARY_PROVIDER_NAMES: "SECONDARY"
+      OIDC_SECONDARY_PROVIDER_NAMES: "secondary"
       OIDC_RP_CLIENT_ID_SECONDARY: "am-storage-service-secondary"
       OIDC_RP_CLIENT_SECRET_SECONDARY: "example-secret-secondary"
       OIDC_OP_AUTHORIZATION_ENDPOINT_SECONDARY: "http://keycloak:8080/realms/secondary/protocol/openid-connect/auth"

tests/storage_service/test_helpers.py

@@ -0,0 +1,111 @@
+import pytest
+from common import helpers
+from django.core.exceptions import ImproperlyConfigured
+
+
+@pytest.mark.parametrize(
+    "environment_variable,expected",
+    [("YES", True), ("foo", False)],
+    ids=["env_var_is_true", "env_var_is_not_true"],
+)
+def test_is_true(environment_variable: str, expected: bool) -> None:
+    assert helpers.is_true(environment_variable) is expected
+
+
+def test_get_env_variable_fails_when_variable_is_not_set() -> None:
+    var_name = "FOO"
+    with pytest.raises(
+        ImproperlyConfigured, match=f"Set the {var_name} environment variable"
+    ):
+        helpers.get_env_variable(var_name)
+
+
+def test_get_env_variable_returns_variable_value_from_environment(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    var_name = "FOO"
+    var_value = "bar"
+    monkeypatch.setenv(var_name, var_value)
+
+    assert helpers.get_env_variable(var_name) == var_value
+
+
+def test_get_oidc_secondary_providers_ignores_provider_if_client_id_and_secret_are_missing(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("OIDC_RP_CLIENT_ID_FOO", "foo-client-id")
+    monkeypatch.setenv("OIDC_RP_CLIENT_SECRET_FOO", "foo-client-secret")
+    monkeypatch.setenv("OIDC_RP_CLIENT_ID_BAR", "bar-client-id")
+    monkeypatch.setenv("OIDC_RP_CLIENT_SECRET_BAZ", "foo-secret")
+
+    assert helpers.get_oidc_secondary_providers(["FOO", "BAR", "BAZ"]) == {
+        "FOO": {
+            "OIDC_OP_AUTHORIZATION_ENDPOINT": "",
+            "OIDC_OP_JWKS_ENDPOINT": "",
+            "OIDC_OP_LOGOUT_ENDPOINT": "",
+            "OIDC_OP_TOKEN_ENDPOINT": "",
+            "OIDC_OP_USER_ENDPOINT": "",
+            "OIDC_RP_CLIENT_ID": "foo-client-id",
+            "OIDC_RP_CLIENT_SECRET": "foo-client-secret",
+        }
+    }
+
+
+def test_get_oidc_secondary_providers_strips_provider_names(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("OIDC_RP_CLIENT_ID_FOO", "foo-client-id")
+    monkeypatch.setenv("OIDC_RP_CLIENT_SECRET_FOO", "foo-client-secret")
+    monkeypatch.setenv("OIDC_RP_CLIENT_ID_BAR", "bar-client-id")
+    monkeypatch.setenv("OIDC_RP_CLIENT_SECRET_BAR", "bar-client-secret")
+
+    assert helpers.get_oidc_secondary_providers(["  FOO", " BAR  "]) == {
+        "FOO": {
+            "OIDC_OP_AUTHORIZATION_ENDPOINT": "",
+            "OIDC_OP_JWKS_ENDPOINT": "",
+            "OIDC_OP_LOGOUT_ENDPOINT": "",
+            "OIDC_OP_TOKEN_ENDPOINT": "",
+            "OIDC_OP_USER_ENDPOINT": "",
+            "OIDC_RP_CLIENT_ID": "foo-client-id",
+            "OIDC_RP_CLIENT_SECRET": "foo-client-secret",
+        },
+        "BAR": {
+            "OIDC_OP_AUTHORIZATION_ENDPOINT": "",
+            "OIDC_OP_JWKS_ENDPOINT": "",
+            "OIDC_OP_LOGOUT_ENDPOINT": "",
+            "OIDC_OP_TOKEN_ENDPOINT": "",
+            "OIDC_OP_USER_ENDPOINT": "",
+            "OIDC_RP_CLIENT_ID": "bar-client-id",
+            "OIDC_RP_CLIENT_SECRET": "bar-client-secret",
+        },
+    }
+
+
+def test_get_oidc_secondary_providers_capitalizes_provider_names(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("OIDC_RP_CLIENT_ID_FOO", "foo-client-id")
+    monkeypatch.setenv("OIDC_RP_CLIENT_SECRET_FOO", "foo-client-secret")
+    monkeypatch.setenv("OIDC_RP_CLIENT_ID_BAR", "bar-client-id")
+    monkeypatch.setenv("OIDC_RP_CLIENT_SECRET_BAR", "bar-client-secret")
+
+    assert helpers.get_oidc_secondary_providers(["fOo", "bar"]) == {
+        "FOO": {
+            "OIDC_OP_AUTHORIZATION_ENDPOINT": "",
+            "OIDC_OP_JWKS_ENDPOINT": "",
+            "OIDC_OP_LOGOUT_ENDPOINT": "",
+            "OIDC_OP_TOKEN_ENDPOINT": "",
+            "OIDC_OP_USER_ENDPOINT": "",
+            "OIDC_RP_CLIENT_ID": "foo-client-id",
+            "OIDC_RP_CLIENT_SECRET": "foo-client-secret",
+        },
+        "BAR": {
+            "OIDC_OP_AUTHORIZATION_ENDPOINT": "",
+            "OIDC_OP_JWKS_ENDPOINT": "",
+            "OIDC_OP_LOGOUT_ENDPOINT": "",
+            "OIDC_OP_TOKEN_ENDPOINT": "",
+            "OIDC_OP_USER_ENDPOINT": "",
+            "OIDC_RP_CLIENT_ID": "bar-client-id",
+            "OIDC_RP_CLIENT_SECRET": "bar-client-secret",
+        },
+    }