fix(oauth): exclude client credentials from body when Authorization header is present and _emit_control_message (#877)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: aldo.gonzalez@airbyte.io <aldo.gonzalez@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py

@@ -100,17 +100,33 @@ def token_has_expired(self) -> bool:
 
     def build_refresh_request_body(self) -> Mapping[str, Any]:
         """
-        Returns the request body to set on the refresh request
+        Returns the request body to set on the refresh request.
 
-        Override to define additional parameters
+        Override to define additional parameters.
+
+        Client credentials (client_id and client_secret) are excluded from the body when
+        refresh_request_headers contains an Authorization header (e.g., Basic auth).
+        This is required by OAuth providers like Gong that expect credentials ONLY in the
+        Authorization header and reject requests that include them in both places.
         """
+        # Check if credentials are being sent via Authorization header
+        headers = self.get_refresh_request_headers()
+        credentials_in_header = headers and "Authorization" in headers
+
+        # Only include client credentials in body if not already in header
+        include_client_credentials = not credentials_in_header
+
         payload: MutableMapping[str, Any] = {
             self.get_grant_type_name(): self.get_grant_type(),
-            self.get_client_id_name(): self.get_client_id(),
-            self.get_client_secret_name(): self.get_client_secret(),
-            self.get_refresh_token_name(): self.get_refresh_token(),
         }
 
+        # Only include client credentials in body if configured to do so and not in header
+        if include_client_credentials:
+            payload[self.get_client_id_name()] = self.get_client_id()
+            payload[self.get_client_secret_name()] = self.get_client_secret()
+
+        payload[self.get_refresh_token_name()] = self.get_refresh_token()
+
         if self.get_scopes():
             payload["scopes"] = self.get_scopes()
 

airbyte_cdk/sources/streams/http/requests_native_auth/oauth.py

@@ -383,27 +383,24 @@ def _emit_control_message(self) -> None:
         """
         Emits a control message based on the connector configuration.
 
-        This method checks if the message repository is not a NoopMessageRepository.
-        If it is not, it emits a message using the message repository. Otherwise,
-        it falls back to emitting the configuration as an Airbyte control message
-        directly to the console for backward compatibility.
+        Control messages for config updates (like refreshed tokens) must be printed directly
+        to stdout so the platform can process them immediately. The message repository is
+        also used to queue the message for any additional processing.
 
         Note:
-            The function `emit_configuration_as_airbyte_control_message` has been deprecated
-            in favor of the package `airbyte_cdk.sources.message`.
-
-        Raises:
-            TypeError: If the argument types are incorrect.
+            The function `emit_configuration_as_airbyte_control_message` prints directly to
+            stdout, which is required for the platform to detect and persist config changes.
         """
-        # FIXME emit_configuration_as_airbyte_control_message as been deprecated in favor of package airbyte_cdk.sources.message
-        # Usually, a class shouldn't care about the implementation details but to keep backward compatibility where we print the
-        # message directly in the console, this is needed
+        # Always emit to stdout so the platform can process the config update immediately.
+        # This is critical for single-use refresh tokens where the new token must be persisted
+        # before subsequent operations try to use the old (now invalid) token.
+        emit_configuration_as_airbyte_control_message(self._connector_config)  # type: ignore[arg-type]
+
+        # Also emit to the message repository for any additional processing (e.g., logging)
         if not isinstance(self._message_repository, NoopMessageRepository):
             self._message_repository.emit_message(
                 create_connector_config_control_message(self._connector_config)  # type: ignore[arg-type]
             )
-        else:
-            emit_configuration_as_airbyte_control_message(self._connector_config)  # type: ignore[arg-type]
 
     @property
     def _message_repository(self) -> MessageRepository:

unit_tests/sources/declarative/auth/test_oauth.py

@@ -111,6 +111,64 @@ def test_refresh_request_headers(self):
         headers = oauth.build_refresh_request_headers()
         assert headers is None
 
+    def test_refresh_request_body_excludes_credentials_when_authorization_header_present(self):
+        """
+        When refresh_request_headers contains an Authorization header (e.g., Basic auth),
+        client_id and client_secret should be excluded from the request body.
+
+        This is required by OAuth providers like Gong that expect credentials ONLY in the
+        Authorization header and reject requests that include them in both places.
+        """
+        oauth = DeclarativeOauth2Authenticator(
+            token_refresh_endpoint="{{ config['refresh_endpoint'] }}",
+            client_id="{{ config['client_id'] }}",
+            client_secret="{{ config['client_secret'] }}",
+            refresh_token="{{ parameters['refresh_token'] }}",
+            config=config,
+            token_expiry_date="{{ config['token_expiry_date'] }}",
+            refresh_request_headers={
+                "Authorization": "Basic {{ [config['client_id'], config['client_secret']] | join(':') | base64encode }}",
+                "Content-Type": "application/x-www-form-urlencoded",
+            },
+            parameters=parameters,
+            grant_type="{{ config['grant_type'] }}",
+        )
+        body = oauth.build_refresh_request_body()
+        expected = {
+            "grant_type": "some_grant_type",
+            "refresh_token": "some_refresh_token",
+        }
+        assert body == expected
+        assert "client_id" not in body
+        assert "client_secret" not in body
+
+    def test_refresh_request_body_includes_credentials_when_no_authorization_header(self):
+        """
+        When refresh_request_headers does NOT contain an Authorization header,
+        client_id and client_secret should be included in the request body (default behavior).
+        """
+        oauth = DeclarativeOauth2Authenticator(
+            token_refresh_endpoint="{{ config['refresh_endpoint'] }}",
+            client_id="{{ config['client_id'] }}",
+            client_secret="{{ config['client_secret'] }}",
+            refresh_token="{{ parameters['refresh_token'] }}",
+            config=config,
+            token_expiry_date="{{ config['token_expiry_date'] }}",
+            refresh_request_headers={
+                "Content-Type": "application/x-www-form-urlencoded",
+            },
+            parameters=parameters,
+            grant_type="{{ config['grant_type'] }}",
+        )
+        body = oauth.build_refresh_request_body()
+        expected = {
+            "grant_type": "some_grant_type",
+            "client_id": "some_client_id",
+            "client_secret": "some_client_secret",
+            "refresh_token": "some_refresh_token",
+        }
+        assert body == expected
+
     def test_refresh_with_encode_config_params(self):
         oauth = DeclarativeOauth2Authenticator(
             token_refresh_endpoint="{{ config['refresh_endpoint'] }}",