feat: add support for content session tokens in snowflake authenticator (#437)

This is now relevant and needed given Snowflake adding support for oauth
federation through an external idp:
https://community.snowflake.com/s/article/Create-External-OAuth-Token-Using-Azure-AD-For-The-OAuth-Client-Itself

Simple change to support both session token types.

Author: mconflitti-pbc

src/posit/connect/external/snowflake.py

@@ -69,11 +69,13 @@ def __init__(
         local_authenticator: Optional[str] = None,
         client: Optional[Client] = None,
         user_session_token: Optional[str] = None,
+        content_session_token: Optional[str] = None,
         audience: Optional[str] = None,
     ):
         self._local_authenticator = local_authenticator
         self._client = client
         self._user_session_token = user_session_token
+        self._content_session_token = content_session_token
         self._audience = audience
 
     @property
@@ -87,16 +89,26 @@ def token(self) -> Optional[str]:
         if is_local():
             return None
 
-        # If the user-session-token wasn't provided and we're running on Connect then we raise an exception.
+        # If a session token wasn't provided and we're running on Connect then we raise an exception.
         # user_session_token is required to impersonate the viewer.
-        if self._user_session_token is None:
-            raise ValueError("The user-session-token is required for viewer authentication.")
+        # content_session_token is required for service account access.
+        if self._user_session_token is None and self._content_session_token is None:
+            raise ValueError(
+                "A user-session-token or content-session-token is required for authentication."
+            )
 
         if self._client is None:
             self._client = Client()
 
-        credentials = self._client.oauth.get_credentials(
-            self._user_session_token,
-            audience=self._audience,
-        )
+        if self._user_session_token is not None:
+            credentials = self._client.oauth.get_credentials(
+                self._user_session_token,
+                audience=self._audience,
+            )
+        else:
+            credentials = self._client.oauth.get_content_credentials(
+                self._content_session_token,
+                audience=self._audience,
+            )
+
         return credentials.get("access_token")

tests/posit/connect/external/test_snowflake.py

@@ -1,5 +1,6 @@
 from unittest.mock import patch
 
+import pytest
 import responses
 
 from posit.connect import Client
@@ -25,6 +26,24 @@ def register_mocks():
         },
     )
 
+    responses.post(
+        "https://connect.example/__api__/v1/oauth/integrations/credentials",
+        match=[
+            responses.matchers.urlencoded_params_matcher(
+                {
+                    "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+                    "subject_token_type": "urn:posit:connect:content-session-token",
+                    "subject_token": "content-token-123",
+                },
+            ),
+        ],
+        json={
+            "access_token": "service-account-access-token",
+            "issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
+            "token_type": "Bearer",
+        },
+    )
+
 
 class TestPositAuthenticator:
     @responses.activate
@@ -42,6 +61,21 @@ def test_posit_authenticator(self):
         assert auth.authenticator == "oauth"
         assert auth.token == "dynamic-viewer-access-token"
 
+    @responses.activate
+    @patch.dict("os.environ", {"RSTUDIO_PRODUCT": "CONNECT"})
+    def test_posit_authenticator_content_token(self):
+        register_mocks()
+
+        client = Client(api_key="12345", url="https://connect.example/")
+        client._ctx.version = None
+        auth = PositAuthenticator(
+            local_authenticator="SNOWFLAKE",
+            content_session_token="content-token-123",
+            client=client,
+        )
+        assert auth.authenticator == "oauth"
+        assert auth.token == "service-account-access-token"
+
     def test_posit_authenticator_fallback(self):
         # local_authenticator is used when the content is running locally
         client = Client(api_key="12345", url="https://connect.example/")
@@ -53,3 +87,31 @@ def test_posit_authenticator_fallback(self):
         )
         assert auth.authenticator == "SNOWFLAKE"
         assert auth.token is None
+
+    def test_posit_authenticator_content_token_fallback(self):
+        # local_authenticator is used when the content is running locally
+        client = Client(api_key="12345", url="https://connect.example/")
+        client._ctx.version = None
+        auth = PositAuthenticator(
+            local_authenticator="SNOWFLAKE",
+            content_session_token="content-token-123",
+            client=client,
+        )
+        assert auth.authenticator == "SNOWFLAKE"
+        assert auth.token is None
+
+    @patch.dict("os.environ", {"RSTUDIO_PRODUCT": "CONNECT"})
+    def test_posit_authenticator_missing_tokens(self):
+        # Should raise an error when running on Connect without any session token
+        client = Client(api_key="12345", url="https://connect.example/")
+        client._ctx.version = None
+        auth = PositAuthenticator(
+            local_authenticator="SNOWFLAKE",
+            client=client,
+        )
+        assert auth.authenticator == "oauth"
+
+        with pytest.raises(
+            ValueError, match="A user-session-token or content-session-token is required"
+        ):
+            _ = auth.token