Prevent OAuth access token from being logged (fastmcp v4)

ads_mcp/utils.py

@@ -41,7 +41,11 @@
 
 logger = logging.getLogger(__name__)
 logging.basicConfig(level=logging.INFO)
-logging.getLogger("httpx").setLevel(logging.WARNING)
+# Prevent fastmcp's HTTP client (httpx2) from logging the OAuth access token
+# fastmcp sends in the request URL to Google's tokeninfo endpoint:
+# https://github.com/PrefectHQ/fastmcp/blob/490049f0f9742922f4af16c937db0b898dc9802b/fastmcp_slim/fastmcp/server/auth/providers/google.py#L116-L120
+# https://github.com/pydantic/httpx2/blob/f2951854442e78cb8f6d2256b7c1d83bd2b77d0b/src/httpx2/httpx2/_client.py#L1923-L1930
+logging.getLogger("httpx2").setLevel(logging.WARNING)
 
 # OAuth scope for the Google Ads API. Google Ads does not publish a separate
 # read-only scope; access is restricted to read methods by the tools this

tests/utils_test.py

@@ -14,6 +14,7 @@
 
 """Test cases for the utils module."""
 
+import logging
 import unittest
 from google.ads.googleads.v25.enums.types.campaign_status import (
     CampaignStatusEnum,
@@ -73,6 +74,29 @@ def test_format_output_value_bare_protobuf(self):
         formatted = utils.format_output_value(fm)
         self.assertEqual(formatted, "foo,bar")
 
+    def test_oauth_access_token_not_logged(self):
+        """Tests that OAuth access token is not logged."""
+        import asyncio
+        import httpx2
+        from fastmcp.server.auth.providers.google import GoogleTokenVerifier
+
+        token = "secret-token"
+        # Include audience (aud) and subject (sub) claims in the mocked tokeninfo response
+        # since they're required by fastmcp's GoogleTokenVerifier's verify_token to make the userinfo request
+        # https://github.com/PrefectHQ/fastmcp/blob/490049f0f9742922f4af16c937db0b898dc9802b/fastmcp_slim/fastmcp/server/auth/providers/google.py#L132-L150
+        transport = httpx2.MockTransport(
+            lambda _: httpx2.Response(200, json={"aud": "aud", "sub": "sub"})
+        )
+        verifier = GoogleTokenVerifier(
+            http_client=httpx2.AsyncClient(transport=transport)
+        )
+
+        with self.assertLogs(level=logging.DEBUG) as logs:
+            asyncio.run(verifier.verify_token(token))
+
+        for line in logs.output:
+            self.assertNotIn(token, line)
+
     def test_prevent_stdio_inheritance(self):
         """Tests that prevent_stdio_inheritance sets stdin to DEVNULL if not specified."""
         import subprocess