Python MGMT SDK: SDK making malformed request for OAuth access tokens in v 3.0.0

.circleci/config.yml

@@ -8,7 +8,8 @@ orbs:
 
 jobs:
   build_and_test:
-    executor: python/default
+    docker:
+      - image: cimg/python:3.10
     steps:
       - checkout
       - python/install-packages:

.github/workflows/python-package.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
 
     steps:
     - uses: actions/checkout@v2

.github/workflows/python.yml

@@ -13,7 +13,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
 
     steps:
       - uses: actions/checkout@v3

okta/http_client.py

@@ -91,28 +91,42 @@ async def send_request(self, request):
         """
         try:
             logger.debug(f"Request: {request}")
-            # Set headers
-            self._default_headers.update(request["headers"])
+            # Create a local copy of headers to avoid mutating shared state
+            request_headers = {**self._default_headers, **request["headers"]}
             # Prepare request parameters
             params = {
                 "method": request["method"],
                 "url": request["url"],
-                "headers": self._default_headers,
+                "headers": request_headers,
             }
             if request["data"]:
                 params["data"] = json.dumps(request["data"])
             elif request["form"]:
-                filename = ""
-                if isinstance(request["form"]["file"], str):
-                    filename = request["form"]["file"].split("/")[-1]
-                data = aiohttp.FormData()
-                data.add_field(
-                    "file",
-                    open(request["form"]["file"], "rb"),
-                    filename=filename,
-                    content_type=self._default_headers["Content-Type"],
-                )
-                params["data"] = data
+                # Check if this is a file upload or form data
+                if "file" in request["form"]:
+                    # File upload
+                    filename = ""
+                    if isinstance(request["form"]["file"], str):
+                        filename = request["form"]["file"].split("/")[-1]
+                    data = aiohttp.FormData()
+                    data.add_field(
+                        "file",
+                        open(request["form"]["file"], "rb"),
+                        filename=filename,
+                        content_type=request_headers["Content-Type"],
+                    )
+                    params["data"] = data
+                else:
+                    # Regular form data (e.g., OAuth client_assertion)
+                    # For application/x-www-form-urlencoded, let aiohttp handle encoding
+                    # by not setting Content-Type header manually
+                    if request_headers.get("Content-Type") == "application/x-www-form-urlencoded":
+                        # Create headers without Content-Type for this request
+                        params["headers"] = {
+                            k: v for k, v in request_headers.items()
+                            if k != "Content-Type"
+                        }
+                    params["data"] = request["form"]
             json_data = request.get("json")
             # empty json param may cause issue, so include it if needed only
             # more details: https://github.com/okta/okta-sdk-python/issues/131

okta/oauth.py

@@ -21,7 +21,6 @@
 """  # noqa: E501
 
 import time
-from urllib.parse import urlencode, quote
 
 from okta.http_client import HTTPClient
 from okta.jwt import JWT
@@ -85,15 +84,14 @@ async def get_access_token(self):
             "client_assertion": jwt,
         }
 
-        encoded_parameters = urlencode(parameters, quote_via=quote)
         org_url = self._config["client"]["orgUrl"]
-        url = f"{org_url}{OAuth.OAUTH_ENDPOINT}?" + encoded_parameters
+        url = f"{org_url}{OAuth.OAUTH_ENDPOINT}"
 
         # Craft request
         oauth_req, err = await self._request_executor.create_request(
             "POST",
             url,
-            form={"client_assertion": jwt},
+            form=parameters,
             headers={
                 "Accept": "application/json",
                 "Content-Type": "application/x-www-form-urlencoded",

openapi/templates/okta/oauth.mustache

@@ -8,7 +8,6 @@
 
 {{>partial_header}}
 import time
-from urllib.parse import urlencode, quote
 from okta.jwt import JWT
 from okta.http_client import HTTPClient
 
@@ -69,14 +68,12 @@ class OAuth:
             'client_assertion': jwt
         }
 
-        encoded_parameters = urlencode(parameters, quote_via=quote)
         org_url = self._config["client"]["orgUrl"]
-        url = f"{org_url}{OAuth.OAUTH_ENDPOINT}?" + \
-            encoded_parameters
+        url = f"{org_url}{OAuth.OAUTH_ENDPOINT}"
 
         # Craft request
         oauth_req, err = await self._request_executor.create_request(
-            "POST", url, form={'client_assertion': jwt}, headers={
+            "POST", url, form=parameters, headers={
                 'Accept': "application/json",
                 'Content-Type': 'application/x-www-form-urlencoded'
             }, oauth=True)

openapi/templates/setup.mustache

@@ -31,7 +31,7 @@ from setuptools import setup, find_packages  # noqa: H301
 # prerequisite: setuptools
 # http://pypi.python.org/pypi/setuptools
 NAME = "okta"
-PYTHON_REQUIRES = ">=3.9"
+PYTHON_REQUIRES = ">=3.10"
 REQUIRES = [
     "aenum >= 3.1.11",
     "aiohttp >= 3.12.14",

setup.py

@@ -31,7 +31,7 @@
 # prerequisite: setuptools
 # http://pypi.python.org/pypi/setuptools
 NAME = "okta"
-PYTHON_REQUIRES = ">=3.9"
+PYTHON_REQUIRES = ">=3.10"
 REQUIRES = [
     "aenum >= 3.1.11",
     "aiohttp >= 3.12.14",

test_header_mutation.py

@@ -0,0 +1,80 @@
+"""
+Simple test to verify OAuth header mutation fix
+"""
+import asyncio
+from okta.http_client import HTTPClient
+
+
+async def test_header_mutation():
+    """Test that sending form data doesn't mutate shared headers"""
+
+    # Initialize HTTPClient with minimal config
+    http_config = {
+        "headers": {
+            "User-Agent": "test-client",
+            "Accept": "application/json"
+        }
+    }
+    http_client = HTTPClient(http_config)
+
+    # Get initial default headers
+    initial_headers = dict(http_client._default_headers)
+    print(f"Initial headers: {initial_headers}")
+
+    # Simulate an OAuth request with form data
+    oauth_request = {
+        "method": "POST",
+        "url": "https://test.okta.com/oauth2/v1/token",
+        "headers": {
+            "Accept": "application/json",
+            "Content-Type": "application/x-www-form-urlencoded"
+        },
+        "data": None,
+        "form": {
+            "grant_type": "client_credentials",
+            "client_assertion": "test_jwt_token"
+        }
+    }
+
+    # This should NOT mutate _default_headers
+    try:
+        # We'll get an error since we're not actually making a request,
+        # but we just want to check header mutation doesn't happen
+        # in the preparation phase
+        _ = await http_client.send_request(oauth_request)
+    except Exception:
+        # Expected to fail, we're just testing header mutation
+        pass
+
+    # Check headers after the request
+    after_headers = dict(http_client._default_headers)
+    print(f"After headers: {after_headers}")
+
+    # Verify headers weren't mutated
+    if initial_headers == after_headers:
+        print("✅ SUCCESS: Headers were not mutated!")
+        print("   Shared state is preserved correctly.")
+        return True
+    else:
+        print("❌ FAILURE: Headers were mutated!")
+        print(f"   Initial: {initial_headers}")
+        print(f"   After:   {after_headers}")
+        added = set(after_headers.keys()) - set(initial_headers.keys())
+        removed = set(initial_headers.keys()) - set(after_headers.keys())
+        if added:
+            print(f"   Added keys: {added}")
+        if removed:
+            print(f"   Removed keys: {removed}")
+        return False
+
+
+if __name__ == '__main__':
+    print("Testing OAuth header mutation fix...")
+    print("=" * 60)
+    result = asyncio.run(test_header_mutation())
+    print("=" * 60)
+    if result:
+        print("All tests passed! ✅")
+    else:
+        print("Tests failed! ❌")
+        exit(1)