- Fixed RSA Key Size Mismatch.

- Fixed Unnecessary Admin URL Removals.
- Fixed OAuth Token Request Behavior Change.
- Added Missing Module - dpop_errors.py.
- Fixed documentation for test File Location.
- Allowed shorter intervals in test/dev environments via constants.py.
- Added Missing Type Hints.
- Addressed Thread Safety Concerns.

Author: BinoyOza-okta

okta/config/config_validator.py

@@ -8,7 +8,7 @@
 # See the License for the specific language governing permissions and limitations under the License.
 # coding: utf-8
 
-from okta.constants import FINDING_OKTA_DOMAIN, REPO_URL
+from okta.constants import FINDING_OKTA_DOMAIN, REPO_URL, MIN_DPOP_KEY_ROTATION_SECONDS
 from okta.error_messages import (
     ERROR_MESSAGE_ORG_URL_MISSING,
     ERROR_MESSAGE_API_TOKEN_DEFAULT,
@@ -166,6 +166,10 @@ def _validate_org_url(self, url: str):
             "-admin.okta.com",
             "-admin.oktapreview.com",
             "-admin.okta-emea.com",
+            "-admin.okta-gov.com",
+            "-admin.okta.mil",
+            "-admin.okta-miltest.com",
+            "-admin.trex-govcloud.com",
         ]
         if any(string in url for string in admin_strings) or "-admin" in url:
             url_errors.append(
@@ -255,9 +259,9 @@ def _validate_dpop_config(self, client):
                 f"dpopKeyRotationInterval must be an integer (seconds), "
                 f"but got {type(rotation_interval).__name__}"
             )
-        elif rotation_interval < 3600:  # Minimum 1 hour
+        elif rotation_interval < MIN_DPOP_KEY_ROTATION_SECONDS:  # Minimum 1 hour
             errors.append(
-                f"dpopKeyRotationInterval must be at least 3600 seconds (1 hour), "
+                f"dpopKeyRotationInterval must be at least {MIN_DPOP_KEY_ROTATION_SECONDS} seconds (1 hour), "
                 f"but got {rotation_interval} seconds. "
                 "Shorter intervals may cause performance issues."
             )

okta/constants.py

@@ -28,3 +28,5 @@
 
 SWA_APP_NAME = "template_swa"
 SWA3_APP_NAME = "template_swa3field"
+
+MIN_DPOP_KEY_ROTATION_SECONDS = 3600

okta/dpop.py

@@ -27,7 +27,7 @@
 import threading
 import time
 import uuid
-from typing import Optional
+from typing import Any, Dict, Optional
 from urllib.parse import urlparse, urlunparse
 
 from Cryptodome.PublicKey import RSA
@@ -58,7 +58,7 @@ class DPoPProofGenerator:
     - Keys are rotated periodically for better security
     """
 
-    def __init__(self, config: dict):
+    def __init__(self, config: Dict[str, Any]) -> None:
         """
         Initialize DPoP proof generator.
 
@@ -67,12 +67,15 @@ def __init__(self, config: dict):
                 - dpopKeyRotationInterval: Key rotation interval in seconds (default: 86400 / 24 hours)
         """
         self._rsa_key: Optional[RSA.RsaKey] = None
-        self._public_jwk: Optional[dict] = None
+        self._public_jwk: Optional[Dict[str, str]] = None
         self._key_created_at: Optional[float] = None
         self._rotation_interval: int = config.get('dpopKeyRotationInterval', 86400)  # 24h default
         self._nonce: Optional[str] = None
-        self._lock = threading.Lock()  # Thread-safe lock for key operations
-        self._active_requests = 0  # Track active requests for safe key rotation
+
+        # Use RLock for reentrant lock support
+        # This allows the same thread to acquire the lock multiple times
+        self._lock: threading.RLock = threading.RLock()
+        self._active_requests: int = 0  # Track active requests for safe key rotation
 
         # Generate initial keys
         self._rotate_keys_internal()
@@ -85,7 +88,7 @@ def _rotate_keys_internal(self) -> None:
 
         Generates a new RSA 2048-bit key pair and exports the public key as JWK.
         """
-        logger.info("Generating new RSA 2048-bit key pair for DPoP")
+        logger.info("Generating new RSA 3072-bit key pair for DPoP")
         self._rsa_key = RSA.generate(3072)
         self._public_jwk = self._export_public_jwk()
         self._key_created_at = time.time()
@@ -125,6 +128,8 @@ def generate_proof_jwt(
         Generate DPoP proof JWT per RFC 9449.
 
         FIX #1: Strips query parameters and fragments from http_url per RFC 9449 Section 4.2.
+        FIX #5 (IMPROVED): Thread-safe key access with proper lock protection to prevent
+        race conditions during key rotation.
 
         Args:
             http_method: HTTP method (GET, POST, etc.)
@@ -146,15 +151,24 @@ def generate_proof_jwt(
             ...     access_token='eyJhbG...'
             ... )
         """
-        # FIX #5: Increment active request counter (thread-safe)
+        # FIX #5 (IMPROVED): Acquire lock and capture key references atomically
+        # This prevents race condition where rotation could happen between
+        # counter increment and key usage
         with self._lock:
             self._active_requests += 1
 
+            # Capture key references while holding lock
+            # This ensures we use consistent key state throughout JWT generation
+            rsa_key = self._rsa_key
+            public_jwk = self._public_jwk
+            key_created_at = self._key_created_at
+            stored_nonce = self._nonce
+
         try:
             # Check if auto-rotation is needed (but don't rotate during active request)
-            if self._should_rotate_keys():
+            if key_created_at and (time.time() - key_created_at) >= self._rotation_interval:
                 logger.warning(
-                    f"DPoP keys are {time.time() - self._key_created_at:.0f}s old, "
+                    f"DPoP keys are {time.time() - key_created_at:.0f}s old, "
                     f"rotation recommended (interval: {self._rotation_interval}s)"
                 )
 
@@ -187,7 +201,7 @@ def generate_proof_jwt(
             }
 
             # Add optional nonce claim (use provided or stored)
-            effective_nonce = nonce or self._nonce
+            effective_nonce = nonce or stored_nonce
             if effective_nonce:
                 claims['nonce'] = effective_nonce
                 logger.debug(f"Added nonce to DPoP proof: {effective_nonce[:8]}...")
@@ -201,13 +215,13 @@ def generate_proof_jwt(
             headers = {
                 'typ': 'dpop+jwt',
                 'alg': 'RS256',
-                'jwk': self._public_jwk
+                'jwk': public_jwk
             }
 
-            # Sign JWT with private key
+            # Sign JWT with private key (using captured reference)
             token = jwt_encode(
                 claims,
-                self._rsa_key.export_key(),
+                rsa_key.export_key(),
                 algorithm='RS256',
                 headers=headers
             )
@@ -221,7 +235,7 @@ def generate_proof_jwt(
             return token
 
         finally:
-            # FIX #5: Decrement active request counter (thread-safe)
+            # FIX #5 (IMPROVED): Decrement counter (thread-safe)
             with self._lock:
                 self._active_requests -= 1
 
@@ -260,7 +274,7 @@ def _compute_access_token_hash(self, access_token: str) -> str:
         logger.debug(f"Computed access token hash: {ath[:16]}...")
         return ath
 
-    def _export_public_jwk(self) -> dict:
+    def _export_public_jwk(self) -> Dict[str, str]:
         """
         Export ONLY public key components as JWK per RFC 7517.
 
@@ -269,7 +283,7 @@ def _export_public_jwk(self) -> dict:
         and MUST NOT contain a private key.
 
         Returns:
-            dict: JWK with only public components (kty, n, e)
+            Dict[str, str]: JWK with only public components (kty, n, e)
 
         Security Note:
             This method uses jwcrypto.export_public() to ensure only public
@@ -331,12 +345,12 @@ def get_nonce(self) -> Optional[str]:
         """
         return self._nonce
 
-    def get_public_jwk(self) -> dict:
+    def get_public_jwk(self) -> Dict[str, str]:
         """
         Get public key in JWK format.
 
         Returns:
-            Copy of the public JWK (kty, n, e)
+            Dict[str, str]: Copy of the public JWK (kty, n, e)
         """
         return self._public_jwk.copy() if self._public_jwk else {}
 

okta/errors/dpop_errors.py

@@ -0,0 +1,69 @@
+"""
+FIX #8: DPoP-specific error messages and handling.
+
+This module provides user-friendly error messages for DPoP-related errors
+returned by the Okta authorization server.
+
+Reference: RFC 9449 Section 7 (Error Handling)
+"""
+
+DPOP_ERROR_MESSAGES = {
+    'invalid_dpop_proof': (
+        'DPoP proof validation failed. The server rejected the DPoP proof JWT. '
+        'Possible causes: invalid signature, incorrect claims, or key mismatch. '
+        'Check that your DPoP keys are correctly generated and the proof JWT '
+        'includes all required claims (jti, htm, htu, iat).'
+    ),
+    'use_dpop_nonce': (
+        'Server requires a nonce in the DPoP proof. '
+        'The SDK will automatically retry with the provided nonce. '
+        'This is normal for the first DPoP request to a server.'
+    ),
+    'invalid_dpop_key_binding': (
+        'Access token is not bound to the DPoP key. '
+        'The access token was obtained with a different key than the one used for this request. '
+        'This may happen if keys were rotated after obtaining the token. '
+        'Try clearing the token cache and obtaining a new token.'
+    ),
+    'invalid_dpop_jkt': (
+        'DPoP JWK thumbprint validation failed. '
+        'The JWK in the DPoP proof does not match the expected thumbprint. '
+        'Ensure you are using the same key pair for all requests.'
+    ),
+    'invalid_request': (
+        'Invalid request. Check your DPoP proof JWT format and claims. '
+        'Ensure the JWT is properly signed and all required claims are present.'
+    ),
+}
+
+
+def get_dpop_error_message(error_code: str) -> str:
+    """
+    Get user-friendly error message for DPoP error code.
+
+    Args:
+        error_code: Error code from OAuth error response
+
+    Returns:
+        User-friendly error message
+    """
+    return DPOP_ERROR_MESSAGES.get(
+        error_code,
+        f'DPoP error: {error_code}. Check Okta logs for details. '
+        f'See RFC 9449 for DPoP specification: https://datatracker.ietf.org/doc/html/rfc9449'
+    )
+
+
+def is_dpop_error(error_code: str) -> bool:
+    """
+    Check if error code is DPoP-related.
+
+    Args:
+        error_code: Error code from OAuth error response
+
+    Returns:
+        True if error is DPoP-related
+    """
+    dpop_keywords = ['dpop', 'nonce', 'jkt', 'key_binding']
+    error_lower = error_code.lower()
+    return any(keyword in error_lower for keyword in dpop_keywords)

okta/oauth.py

@@ -22,6 +22,7 @@
 
 import logging
 import time
+from typing import Any, Dict, Optional, Tuple
 from urllib.parse import urlencode, quote
 
 from okta.http_client import HTTPClient
@@ -37,22 +38,23 @@ class OAuth:
 
     OAUTH_ENDPOINT = "/oauth2/v1/token"
 
-    def __init__(self, request_executor, config):
+    def __init__(self, request_executor: Any, config: Dict[str, Any]) -> None:
         self._request_executor = request_executor
         self._config = config
-        self._access_token = None
-        self._token_type = "Bearer"  # FIX #4: Default token type
+        self._access_token: Optional[str] = None
+        self._token_type: str = "Bearer"  # FIX #4: Default token type
+        self._access_token_expiry_time: Optional[int] = None
 
         # FIX #3, #7: Initialize DPoP if enabled
-        self._dpop_enabled = config["client"].get("dpopEnabled", False)
-        self._dpop_generator = None
+        self._dpop_enabled: bool = config["client"].get("dpopEnabled", False)
+        self._dpop_generator: Optional[Any] = None
 
         if self._dpop_enabled:
             from okta.dpop import DPoPProofGenerator
             self._dpop_generator = DPoPProofGenerator(config["client"])
             logger.info("DPoP authentication enabled")
 
-    def get_JWT(self):
+    def get_JWT(self) -> str:
         """
         Generates JWT using client configuration
 
@@ -67,7 +69,7 @@ def get_JWT(self):
 
         return JWT.create_token(org_url, client_id, private_key, kid)
 
-    async def get_access_token(self):
+    async def get_access_token(self) -> Tuple[Optional[str], str, Optional[Exception]]:
         """
         Retrieves or generates the OAuth access token for the Okta Client.
         Supports both Bearer and DPoP token types.
@@ -120,7 +122,7 @@ async def get_access_token(self):
         oauth_req, err = await self._request_executor.create_request(
             "POST",
             url,
-            form={},  # Parameters are already in the URL
+            form={"client_assertion": jwt},
             headers=headers,
             oauth=True,
         )
@@ -221,7 +223,7 @@ async def get_access_token(self):
 
         return (access_token, token_type, None)
 
-    def clear_access_token(self):
+    def clear_access_token(self) -> None:
         """
         Clear currently used OAuth access token, probably expired.
         FIX #4: Also clears token type.
@@ -233,6 +235,6 @@ def clear_access_token(self):
         self._request_executor._default_headers.pop("Authorization", None)
         self._access_token_expiry_time = None
 
-    def get_dpop_generator(self):
+    def get_dpop_generator(self) -> Optional[Any]:
         """Get DPoP generator instance."""
         return self._dpop_generator

tests/test_dpop.py

@@ -39,7 +39,7 @@ def test_initialization(self):
     def test_key_generation(self):
         """Test RSA 2048-bit key generation."""
         # Key should be RSA
-        self.assertEqual(self.generator._rsa_key.size_in_bits(), 2048)
+        self.assertEqual(self.generator._rsa_key.size_in_bits(), 3072)
 
         # Should have both public and private components
         self.assertTrue(self.generator._rsa_key.has_private())