Format and filter session_required_* query parameters

changelog.d/20260909_165315_kurtmckee_update_get_authorize_url_issue_1425.rst

@@ -0,0 +1,5 @@
+Fixed
+-----
+
+- Format and filter ``session_required_*`` query parameters
+  when generating authorization URLs. (:pr:`NUMBER`)

src/globus_sdk/services/auth/client/base_login_client.py

@@ -7,7 +7,6 @@
 from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
 
 from globus_sdk import client, exc
-from globus_sdk._internal.remarshal import commajoin
 from globus_sdk._missing import MISSING, MissingType
 from globus_sdk.authorizers import GlobusAuthorizer, NullAuthorizer
 from globus_sdk.response import GlobusHTTPResponse
@@ -195,9 +194,9 @@ def oauth2_get_authorize_url(
                 "AuthClient to resolve"
             )
         query_params = {
-            "session_required_identities": commajoin(session_required_identities),
-            "session_required_single_domain": commajoin(session_required_single_domain),
-            "session_required_policies": commajoin(session_required_policies),
+            "session_required_identities": session_required_identities,
+            "session_required_single_domain": session_required_single_domain,
+            "session_required_policies": session_required_policies,
             "session_required_mfa": session_required_mfa,
             "session_message": session_message,
             "prompt": prompt,

src/globus_sdk/services/auth/flow_managers/authorization_code.py

@@ -2,10 +2,8 @@
 
 import logging
 import typing as t
-import urllib.parse
 
 from globus_sdk._internal.utils import slash_join
-from globus_sdk._missing import filter_missing
 from globus_sdk.scopes import Scope, ScopeParser
 
 from ..response import OAuthAuthorizationCodeResponse
@@ -75,7 +73,7 @@ def __init__(
 
     def get_authorize_url(self, query_params: dict[str, t.Any] | None = None) -> str:
         """
-        Start a Authorization Code flow by getting the authorization URL to
+        Start an Authorization Code flow by getting the authorization URL to
         which users should be sent.
 
         :param query_params: Additional parameters to include in the authorize URL.
@@ -86,24 +84,22 @@ def get_authorize_url(self, query_params: dict[str, t.Any] | None = None) -> str
         either to your provided ``redirect_uri`` or to the default location,
         with the ``auth_code`` embedded in a query parameter.
         """
-        authorize_base_url = slash_join(
-            self.auth_client.base_url, "/v2/oauth2/authorize"
-        )
-        log.debug(f"Building authorization URI. Base URL: {authorize_base_url}")
-        log.debug(f"query_params={query_params}")
 
-        params = {
+        base_url = slash_join(self.auth_client.base_url, "/v2/oauth2/authorize")
+        base_query_params = {
             "client_id": self.client_id,
             "redirect_uri": self.redirect_uri,
             "scope": self.requested_scopes,
             "state": self.state,
             "response_type": "code",
             "access_type": (self.refresh_tokens and "offline") or "online",
-            **(query_params or {}),
         }
-        params = filter_missing(params)
-        encoded_params = urllib.parse.urlencode(params)
-        return f"{authorize_base_url}?{encoded_params}"
+
+        return super()._get_authorize_url(
+            base_url=base_url,
+            base_query_params=base_query_params,
+            query_params=query_params,
+        )
 
     def exchange_code_for_tokens(
         self, auth_code: str

src/globus_sdk/services/auth/flow_managers/base.py

@@ -1,10 +1,17 @@
 from __future__ import annotations
 
 import abc
+import logging
 import typing as t
+import urllib.parse
+
+from globus_sdk._internal.remarshal import commajoin
+from globus_sdk._missing import filter_missing
 
 from ..response import OAuthAuthorizationCodeResponse
 
+log = logging.getLogger(__name__)
+
 
 class GlobusOAuthFlowManager(abc.ABC):
     """
@@ -40,6 +47,52 @@ def get_authorize_url(self, query_params: dict[str, t.Any] | None = None) -> str
             as query params on the URL.
         """
 
+    @staticmethod
+    def _get_authorize_url(
+        *,
+        base_url: str,
+        base_query_params: dict[str, t.Any],
+        query_params: dict[str, t.Any] | None,
+    ) -> str:
+        """
+        Get an authorize URL.
+
+        Query parameters may be formatted or excluded to meet known requirements.
+
+        :param base_url:
+            The base URL, which may include a path.
+        :param base_query_params:
+            The base query parameters that are provided by the subclass.
+        :param query_params:
+            Query parameters that are provided by callers.
+            These will always override *base_query_params*
+            but are still subject to format and exclusion requirements.
+        """
+
+        log.debug(f"Building authorization URI. Base URL: {base_url}")
+        log.debug(f"query_params={query_params}")
+
+        params = {
+            **base_query_params,
+            **(query_params or {}),
+        }
+        params = filter_missing(params)
+
+        # Format well-known keys, if they have truth-y values.
+        comma_joined_keys = {
+            "session_required_identities",
+            "session_required_single_domain",
+            "session_required_policies",
+        }
+        for key in comma_joined_keys:
+            # Pop the value and only re-set it if it's truth-y.
+            value = params.pop(key, None)
+            if value:
+                params[key] = commajoin(value)
+
+        encoded_params = urllib.parse.urlencode(params)
+        return f"{base_url}?{encoded_params}"
+
     @abc.abstractmethod
     def exchange_code_for_tokens(
         self, auth_code: str

src/globus_sdk/services/auth/flow_managers/native_app.py

@@ -6,10 +6,9 @@
 import os
 import re
 import typing as t
-import urllib.parse
 
 from globus_sdk._internal.utils import slash_join
-from globus_sdk._missing import MISSING, MissingType, filter_missing
+from globus_sdk._missing import MISSING, MissingType
 from globus_sdk.exc import GlobusSDKUsageError
 from globus_sdk.scopes import Scope, ScopeParser
 
@@ -168,28 +167,25 @@ def get_authorize_url(self, query_params: dict[str, t.Any] | None = None) -> str
         either to your provided ``redirect_uri`` or to the default location,
         with the ``auth_code`` embedded in a query parameter.
         """
-        authorize_base_url = slash_join(
-            self.auth_client.base_url, "/v2/oauth2/authorize"
-        )
-        log.debug(f"Building authorization URI. Base URL: {authorize_base_url}")
-        log.debug(f"query_params={query_params}")
 
-        params = {
+        base_url = slash_join(self.auth_client.base_url, "/v2/oauth2/authorize")
+        base_query_params = {
             "client_id": self.client_id,
             "redirect_uri": self.redirect_uri,
             "scope": self.requested_scopes,
             "state": self.state,
             "response_type": "code",
+            "access_type": (self.refresh_tokens and "offline") or "online",
             "code_challenge": self.challenge,
             "code_challenge_method": "S256",
-            "access_type": (self.refresh_tokens and "offline") or "online",
             "prefill_named_grant": self.prefill_named_grant,
-            **(query_params or {}),
         }
-        params = filter_missing(params)
 
-        encoded_params = urllib.parse.urlencode(params)
-        return f"{authorize_base_url}?{encoded_params}"
+        return super()._get_authorize_url(
+            base_url=base_url,
+            base_query_params=base_query_params,
+            query_params=query_params,
+        )
 
     def exchange_code_for_tokens(
         self, auth_code: str

tests/unit/helpers/test_auth_flow_managers.py

@@ -6,6 +6,7 @@
 import pytest
 
 import globus_sdk
+from globus_sdk import MISSING
 from globus_sdk.scopes import TransferScopes
 from globus_sdk.services.auth.flow_managers.authorization_code import (
     GlobusAuthorizationCodeFlowManager,
@@ -69,13 +70,82 @@ def test_get_authorize_url_for_authorization_code():
         requested_scopes=TransferScopes.all,
     )
 
-    silly_string = "ANANAS_IS_PINEAPPLE_BUT_BANANE_IS_BANANA"
+    value = ["apples", "bananas"]
 
     authorize_url = flow_manager.get_authorize_url()
     assert authorize_url.startswith("https://auth.globus.org")
-    assert silly_string not in authorize_url
+    assert "session_required_identities=" not in authorize_url
 
-    silly_authorize_url = flow_manager.get_authorize_url(
-        query_params={"silly_string": silly_string}
+    fruity_authorize_url = flow_manager.get_authorize_url(
+        query_params={"session_required_identities": value}
     )
-    assert silly_string in silly_authorize_url
+    assert "session_required_identities=apples%2Cbananas" in fruity_authorize_url
+
+
+@pytest.mark.parametrize("parameter", ("base_query_params", "query_params"))
+@pytest.mark.parametrize(
+    "key",
+    (
+        "session_required_identities",
+        "session_required_single_domain",
+        "session_required_policies",
+    ),
+)
+def test_get_authorize_url_formatting(parameter, key):
+    """
+    Verify 'session_required_*' values are comma-joined.
+
+    Prioritization of *query_params* over *base_query_params* is also tested
+    by confirming that the "wrong-value" in *base_query_params* is overridden
+    when the *key* is set in *query_params*.
+    """
+
+    # Arrange
+    parameters = {
+        "base_url": "https://auth.globus.org/",
+        "base_query_params": {
+            "session_required_identities": "wrong-value",
+            "session_required_single_domain": "wrong-value",
+            "session_required_policies": "wrong-value",
+        },
+        "query_params": {},
+    }
+    parameters[parameter][key] = ["correct", "value"]
+
+    # Act
+    url = GlobusAuthorizationCodeFlowManager._get_authorize_url(**parameters)
+
+    # Assert
+    assert f"{key}=correct%2Cvalue" in url
+
+
+@pytest.mark.parametrize("parameter", ("base_query_params", "query_params"))
+@pytest.mark.parametrize(
+    "key",
+    (
+        "session_required_identities",
+        "session_required_single_domain",
+        "session_required_policies",
+    ),
+)
+@pytest.mark.parametrize("value", (MISSING, None, []))
+def test_get_authorize_url_exclusions(key, parameter, value):
+    """Verify false-y 'session_required_*' values not serialized."""
+
+    # Arrange
+    parameters = {
+        "base_url": "https://auth.globus.org/",
+        "base_query_params": {
+            "session_required_identities": "wrong-value",
+            "session_required_single_domain": "wrong-value",
+            "session_required_policies": "wrong-value",
+        },
+        "query_params": {},
+    }
+    parameters[parameter][key] = value
+
+    # Act
+    url = GlobusAuthorizationCodeFlowManager._get_authorize_url(**parameters)
+
+    # Assert
+    assert f"{key}=" not in url