added external provider for connect api

Author: mconflitti-pbc

src/posit/connect/external/connect_api.py

@@ -0,0 +1,78 @@
+"""Connect API integration.
+
+Connect API key exchange implementation which supports interacting with Posit OAuth integrations on Connect.
+
+Notes
+-----
+The APIs in this module are provided as a convenience and are subject to breaking changes.
+"""
+
+from typing_extensions import Optional
+
+from ..oauth.oauth import API_KEY_TOKEN_TYPE
+
+from ..client import Client
+from .external import is_local
+
+
+class ConnectAPIKeyProvider:
+    """
+    Provider for exchanging Connect User Session Token for Connect API Key that acts on behalf of the user.
+    This is an ephemeral API key that adheres to typical ephemeral API key clean up processes.
+
+    Examples
+    --------
+    ```python
+    from shiny import App, render, ui, reactive, req
+    from posit.connect import Client
+    from posit.connect.external.connect_api import ConnectAPIKeyProvider
+
+    client = Client()
+
+    app_ui = ui.page_fixed(
+        ui.h1("My Shiny App"),
+        # ...
+    )
+
+    def server(input, output, session):
+        user_session_token = session.http_conn.headers.get("Posit-Connect-User-Session-Token")
+        provider = ConnectAPIKeyProvider(client, user_session_token)
+        viewer_api_key = provider.viewer_key
+
+        # your app logic...
+
+    app = App(app_ui, server)
+    ```
+    """
+
+    def __init__(
+        self,
+        client: Optional[Client] = None,
+        user_session_token: Optional[str] = None,
+    ):
+        self._client = client
+        self._user_session_token = user_session_token
+
+    @property
+    def viewer_key(self) -> Optional[str]:
+        """
+        The viewer key is retrieved through an OAuth exchange process using the user session token.
+        The issued API key is associated with the viewer of your app and can be used on their behalf
+        to interact with the Connect API.
+        """
+        if is_local():
+            return None
+
+        # If the user-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 API key authorization.")
+
+        if self._client is None:
+            self._client = Client()
+
+        credentials = self._client.oauth.get_credentials(
+            user_session_token=self._user_session_token,
+            requested_token_type=API_KEY_TOKEN_TYPE,
+        )
+        return credentials.get("access_token")

src/posit/connect/oauth/oauth.py

@@ -14,6 +14,7 @@
 GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"
 USER_SESSION_TOKEN_TYPE = "urn:posit:connect:user-session-token"
 CONTENT_SESSION_TOKEN_TYPE = "urn:posit:connect:content-session-token"
+API_KEY_TOKEN_TYPE = "urn:posit:connect:api-key"
 
 
 def _get_content_session_token() -> str:
@@ -51,14 +52,16 @@ def integrations(self):
     def sessions(self):
         return Sessions(self._ctx)
 
-    def get_credentials(self, user_session_token: Optional[str] = None) -> Credentials:
+    def get_credentials(self, user_session_token: Optional[str] = None, requested_token_type: Optional[str] = None) -> Credentials:
         """Perform an oauth credential exchange with a user-session-token."""
         # craft a credential exchange request
         data = {}
         data["grant_type"] = GRANT_TYPE
         data["subject_token_type"] = USER_SESSION_TOKEN_TYPE
         if user_session_token:
             data["subject_token"] = user_session_token
+        if requested_token_type:
+            data["requested_token_type"] = requested_token_type
 
         response = self._ctx.client.post(self._path, data=data)
         return Credentials(**response.json())

tests/posit/connect/oauth/test_oauth.py

@@ -41,6 +41,36 @@ def test_get_credentials(self):
         assert "access_token" in creds
         assert creds["access_token"] == "viewer-token"
 
+    @responses.activate
+    def test_get_credentials_api_key(self):
+        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:user-session-token",
+                        "subject_token": "cit",
+                        "requested_token_type": "urn:posit:connect:api-key"
+                    },
+                ),
+            ],
+            json={
+                "access_token": "viewer-api-key",
+                "issued_token_type": "urn:posit:connect:api-key",
+                "token_type": "Key",
+            },
+        )
+        c = Client(api_key="12345", url="https://connect.example/")
+        c._ctx.version = None
+        creds = c.oauth.get_credentials("cit")
+        assert "access_token" in creds
+        assert creds["access_token"] == "viewer-api-key"
+        assert "issued_token_type" in creds
+        assert creds["issued_token_type"] == "urn:posit:connect:api-key"
+        assert "token_type" in creds
+        assert creds["token_type"] == "Key"
+
     @responses.activate
     def test_get_content_credentials(self):
         responses.post(