feat: Implement token revocation in STS client and add revoke() metho…

google/auth/external_account_authorized_user.py

@@ -321,6 +321,30 @@ def _build_trust_boundary_lookup_url(self):
             universe_domain=self._universe_domain, pool_id=pool_id
         )
 
+    def revoke(self, request):
+        """Revokes the refresh token.
+
+        Args:
+            request (google.auth.transport.Request): The object used to make
+                HTTP requests.
+
+        Raises:
+            google.auth.exceptions.OAuthError: If the token could not be
+                revoked.
+        """
+        if not self._revoke_url or not self._refresh_token_val:
+            raise exceptions.OAuthError(
+                "The credentials do not contain the necessary fields to "
+                "revoke the refresh token. You must specify revoke_url and "
+                "refresh_token."
+            )
+
+        self._sts_client.revoke_token(
+            request, self._refresh_token_val, "refresh_token", self._revoke_url
+        )
+        self.token = None
+        self._refresh_token = None
+
     @_helpers.copy_docstring(credentials.Credentials)
     def get_cred_info(self):
         if self._cred_file_path:

google/oauth2/sts.py

@@ -57,7 +57,7 @@ def __init__(self, token_exchange_endpoint, client_authentication=None):
         super(Client, self).__init__(client_authentication)
         self._token_exchange_endpoint = token_exchange_endpoint
 
-    def _make_request(self, request, headers, request_body):
+    def _make_request(self, request, headers, request_body, url=None):
         # Initialize request headers.
         request_headers = _URLENCODED_HEADERS.copy()
 
@@ -69,9 +69,12 @@ def _make_request(self, request, headers, request_body):
         # Apply OAuth client authentication.
         self.apply_client_authentication_options(request_headers, request_body)
 
+        # Use default token exchange endpoint if no url is provided.
+        url = url or self._token_exchange_endpoint
+
         # Execute request.
         response = request(
-            url=self._token_exchange_endpoint,
+            url=url,
             method="POST",
             headers=request_headers,
             body=urllib.parse.urlencode(request_body).encode("utf-8"),
@@ -87,10 +90,12 @@ def _make_request(self, request, headers, request_body):
         if response.status != http_client.OK:
             utils.handle_error_response(response_body)
 
-        response_data = json.loads(response_body)
+        # A successful token revocation returns an empty response body.
+        if not response_body:
+            return {}
 
-        # Return successful response.
-        return response_data
+        # Other successful responses should be valid JSON.
+        return json.loads(response_body)
 
     def exchange_token(
         self,
@@ -174,3 +179,23 @@ def refresh_token(self, request, refresh_token):
             None,
             {"grant_type": "refresh_token", "refresh_token": refresh_token},
         )
+
+    def revoke_token(self, request, token, token_type_hint, revoke_url):
+        """Revokes the provided token based on the RFC7009 spec.
+
+        Args:
+            request (google.auth.transport.Request): A callable used to make
+                HTTP requests.
+            token (str): The OAuth 2.0 token to revoke.
+            token_type_hint (str): Hint for the type of token being revoked.
+            revoke_url (str): The STS endpoint URL for revoking tokens.
+
+        Raises:
+            google.auth.exceptions.OAuthError: If the token revocation endpoint
+                returned an error.
+        """
+        request_body = {"token": token}
+        if token_type_hint:
+            request_body["token_type_hint"] = token_type_hint
+
+        return self._make_request(request, None, request_body, revoke_url)

tests/oauth2/test_sts.py

@@ -41,6 +41,9 @@ class TestStsClient(object):
     ACTOR_TOKEN = "HEADER.ACTOR_TOKEN_PAYLOAD.SIGNATURE"
     ACTOR_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:jwt"
     TOKEN_EXCHANGE_ENDPOINT = "https://example.com/token.oauth2"
+    REVOKE_URL = "https://example.com/revoke.oauth2"
+    TOKEN_TO_REVOKE = "TOKEN_TO_REVOKE"
+    TOKEN_TYPE_HINT = "refresh_token"
     ADDON_HEADERS = {"x-client-version": "0.1.2"}
     ADDON_OPTIONS = {"additional": {"non-standard": ["options"], "other": "some-value"}}
     SUCCESS_RESPONSE = {
@@ -72,21 +75,24 @@ def make_client(cls, client_auth=None):
         return sts.Client(cls.TOKEN_EXCHANGE_ENDPOINT, client_auth)
 
     @classmethod
-    def make_mock_request(cls, data, status=http_client.OK):
+    def make_mock_request(cls, data, status=http_client.OK, use_json=True):
         response = mock.create_autospec(transport.Response, instance=True)
         response.status = status
-        response.data = json.dumps(data).encode("utf-8")
+        if use_json:
+            response.data = json.dumps(data).encode("utf-8")
+        else:
+            response.data = data.encode("utf-8")
 
         request = mock.create_autospec(transport.Request)
         request.return_value = response
 
         return request
 
     @classmethod
-    def assert_request_kwargs(cls, request_kwargs, headers, request_data):
-        """Asserts the request was called with the expected parameters.
-        """
-        assert request_kwargs["url"] == cls.TOKEN_EXCHANGE_ENDPOINT
+    def assert_request_kwargs(cls, request_kwargs, headers, request_data, url=None):
+        """Asserts the request was called with the expected parameters."""
+        url = url or cls.TOKEN_EXCHANGE_ENDPOINT
+        assert request_kwargs["url"] == url
         assert request_kwargs["method"] == "POST"
         assert request_kwargs["headers"] == headers
         assert request_kwargs["body"] is not None
@@ -447,6 +453,63 @@ def test_refresh_token_failure(self):
             r"Error code invalid_request: Invalid subject token - https://tools.ietf.org/html/rfc6749"
         )
 
+    def test_revoke_token_success(self):
+        """Test revoke token with successful response."""
+        client = self.make_client(self.CLIENT_AUTH_BASIC)
+        request = self.make_mock_request(data="", status=http_client.OK, use_json=False)
+
+        response = client.revoke_token(
+            request, self.TOKEN_TO_REVOKE, self.TOKEN_TYPE_HINT, self.REVOKE_URL
+        )
+
+        headers = {
+            "Authorization": "Basic {}".format(BASIC_AUTH_ENCODING),
+            "Content-Type": "application/x-www-form-urlencoded",
+        }
+        request_data = {
+            "token": self.TOKEN_TO_REVOKE,
+            "token_type_hint": self.TOKEN_TYPE_HINT,
+        }
+        self.assert_request_kwargs(
+            request.call_args[1], headers, request_data, url=self.REVOKE_URL
+        )
+        assert response == {}
+
+    def test_revoke_token_success_no_hint(self):
+        """Test revoke token with successful response."""
+        client = self.make_client(self.CLIENT_AUTH_BASIC)
+        request = self.make_mock_request(data="", status=http_client.OK, use_json=False)
+
+        response = client.revoke_token(
+            request, self.TOKEN_TO_REVOKE, None, self.REVOKE_URL
+        )
+
+        headers = {
+            "Authorization": "Basic {}".format(BASIC_AUTH_ENCODING),
+            "Content-Type": "application/x-www-form-urlencoded",
+        }
+        request_data = {"token": self.TOKEN_TO_REVOKE}
+        self.assert_request_kwargs(
+            request.call_args[1], headers, request_data, url=self.REVOKE_URL
+        )
+        assert response == {}
+
+    def test_revoke_token_failure(self):
+        """Test revoke token with failure response."""
+        client = self.make_client(self.CLIENT_AUTH_BASIC)
+        request = self.make_mock_request(
+            status=http_client.BAD_REQUEST, data=self.ERROR_RESPONSE
+        )
+
+        with pytest.raises(exceptions.OAuthError) as excinfo:
+            client.revoke_token(
+                request, self.TOKEN_TO_REVOKE, self.TOKEN_TYPE_HINT, self.REVOKE_URL
+            )
+
+        assert excinfo.match(
+            r"Error code invalid_request: Invalid subject token - https://tools.ietf.org/html/rfc6749"
+        )
+
     def test__make_request_success(self):
         """Test base method with successful response."""
         client = self.make_client(self.CLIENT_AUTH_BASIC)
@@ -478,3 +541,12 @@ def test_make_request_failure(self):
         assert excinfo.match(
             r"Error code invalid_request: Invalid subject token - https://tools.ietf.org/html/rfc6749"
         )
+
+    def test__make_request_empty_response(self):
+        """Test _make_request with a successful but empty response body."""
+        client = self.make_client()
+        request = self.make_mock_request(data="", status=http_client.OK, use_json=False)
+
+        response = client._make_request(request, {}, {})
+
+        assert response == {}

tests/test_external_account_authorized_user.py

@@ -349,6 +349,50 @@ def test_refresh_without_client_secret(self):
 
         request.assert_not_called()
 
+    def test_revoke_auth_success(self):
+        request = self.make_mock_request(status=http_client.OK, data={})
+        creds = self.make_credentials(revoke_url=REVOKE_URL)
+
+        creds.revoke(request)
+
+        request.assert_called_once_with(
+            url=REVOKE_URL,
+            method="POST",
+            headers={
+                "Content-Type": "application/x-www-form-urlencoded",
+                "Authorization": "Basic " + BASIC_AUTH_ENCODING,
+            },
+            body=("token=" + REFRESH_TOKEN + "&token_type_hint=refresh_token").encode(
+                "utf-8"
+            ),
+        )
+        assert creds.token is None
+        assert creds._refresh_token is None
+
+    def test_revoke_without_revoke_url(self):
+        request = self.make_mock_request()
+        creds = self.make_credentials(token=ACCESS_TOKEN)
+
+        with pytest.raises(exceptions.OAuthError) as excinfo:
+            creds.revoke(request)
+
+        assert excinfo.match(
+            r"The credentials do not contain the necessary fields to revoke the refresh token. You must specify revoke_url and refresh_token."
+        )
+
+    def test_revoke_without_refresh_token(self):
+        request = self.make_mock_request()
+        creds = self.make_credentials(
+            refresh_token=None, token=ACCESS_TOKEN, revoke_url=REVOKE_URL
+        )
+
+        with pytest.raises(exceptions.OAuthError) as excinfo:
+            creds.revoke(request)
+
+        assert excinfo.match(
+            r"The credentials do not contain the necessary fields to revoke the refresh token. You must specify revoke_url and refresh_token."
+        )
+
     def test_info(self):
         creds = self.make_credentials()
         info = creds.info