added aws helper for content session token exchange

Author: mconflitti-pbc

src/posit/connect/external/aws.py

@@ -30,6 +30,7 @@ def get_aws_credentials(client: Client, user_session_token: str) -> Dict[str, st
     client = Client()
     session_token = session.http_conn.headers.get("Posit-Connect-User-Session-Token")
     credentials = get_aws_credentials(client, user_session_token)
+    aws_session_expiration = credentials["expiration"]
     aws_session = boto3.Session(
         aws_access_key_id=credentials["accessKeyId"],
         aws_secret_access_key=credentials["secretAccessKey"],
@@ -85,6 +86,7 @@ def get_aws_content_credentials(
 
     client = Client()
     credentials = get_aws_content_credentials(client)
+    session_expiration = credentials["expiration"]
     session = boto3.Session(
         aws_access_key_id=credentials["accessKeyId"],
         aws_secret_access_key=credentials["secretAccessKey"],

tests/posit/connect/external/test_aws.py

@@ -2,7 +2,11 @@
 import responses
 
 from posit.connect import Client
-from posit.connect.external.aws import get_aws_credentials
+from posit.connect.external.aws import (
+    _decode_access_token,
+    get_aws_content_credentials,
+    get_aws_credentials,
+)
 
 aws_creds = {
     "accessKeyId": "abc123",
@@ -66,3 +70,59 @@ def test_get_aws_credentials_no_token(self):
             get_aws_credentials(c, "cit")
 
         assert e.match("No access token found in credentials")
+
+    @responses.activate
+    def test_get_aws_content_credentials(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:content-session-token",
+                        "subject_token": "cit",
+                        "requested_token_type": "urn:ietf:params:aws:token-type:credentials",
+                    }
+                )
+            ],
+            json={
+                "access_token": encoded_aws_creds,
+                "issued_token_type": "urn:ietf:params:aws:token-type:credentials",
+                "token_type": "aws_credentials",
+            },
+        )
+
+        c = Client(api_key="12345", url="https://connect.example/")
+        c._ctx.version = None
+        response = get_aws_content_credentials(c, "cit")
+
+        assert response == aws_creds
+
+    @responses.activate
+    def test_get_aws_content_credentials_no_token(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:content-session-token",
+                        "subject_token": "cit",
+                        "requested_token_type": "urn:ietf:params:aws:token-type:credentials",
+                    }
+                )
+            ],
+            json={},
+        )
+
+        c = Client(api_key="12345", url="https://connect.example/")
+        c._ctx.version = None
+
+        with pytest.raises(ValueError) as e:
+            get_aws_content_credentials(c, "cit")
+
+        assert e.match("No access token found in credentials")
+
+    def test_decode_access_token(self):
+        decoded_creds = _decode_access_token(encoded_aws_creds)
+        assert decoded_creds == aws_creds

tests/posit/connect/oauth/test_oauth.py

@@ -26,6 +26,7 @@ def test_get_credentials(self):
                         "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
                         "subject_token_type": "urn:posit:connect:user-session-token",
                         "subject_token": "cit",
+                        # no requested token type set
                     },
                 ),
             ],
@@ -41,7 +42,7 @@ def test_get_credentials(self):
         assert creds.get("access_token") == "viewer-token"
 
     @responses.activate
-    def test_get_credentials_api_key(self):
+    def test_get_credentials_with_requested_token_type(self):
         responses.post(
             "https://connect.example/__api__/v1/oauth/integrations/credentials",
             match=[
@@ -68,34 +69,32 @@ def test_get_credentials_api_key(self):
         assert creds.get("token_type") == "Key"
 
     @responses.activate
-    def test_get_credentials_aws(self):
+    def test_get_content_credentials(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_type": "urn:posit:connect:content-session-token",
                         "subject_token": "cit",
-                        "requested_token_type": "urn:ietf:params:aws:token-type:credentials",
+                        # no requested token type set
                     },
                 ),
             ],
             json={
-                "access_token": "encoded-aws-creds",
-                "issued_token_type": "urn:ietf:params:aws:token-type:credentials",
-                "token_type": "aws_credentials",
+                "access_token": "content-token",
+                "issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
+                "token_type": "Bearer",
             },
         )
         c = Client(api_key="12345", url="https://connect.example/")
         c._ctx.version = None
-        creds = c.oauth.get_credentials("cit", OAuthTokenType.AWS_CREDENTIALS)
-        assert creds.get("access_token") == "encoded-aws-creds"
-        assert creds.get("issued_token_type") == "urn:ietf:params:aws:token-type:credentials"
-        assert creds.get("token_type") == "aws_credentials"
+        creds = c.oauth.get_content_credentials("cit")
+        assert creds.get("access_token") == "content-token"
 
     @responses.activate
-    def test_get_content_credentials(self):
+    def test_get_content_credentials_with_requested_token_type(self):
         responses.post(
             "https://connect.example/__api__/v1/oauth/integrations/credentials",
             match=[
@@ -104,19 +103,22 @@ def test_get_content_credentials(self):
                         "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
                         "subject_token_type": "urn:posit:connect:content-session-token",
                         "subject_token": "cit",
+                        "requested_token_type": "urn:ietf:params:aws:token-type:credentials",
                     },
                 ),
             ],
             json={
-                "access_token": "content-token",
-                "issued_token_type": "urn:ietf:params:oauth:token-type:access_token",
-                "token_type": "Bearer",
+                "access_token": "encoded-aws-creds",
+                "issued_token_type": "urn:ietf:params:aws:token-type:credentials",
+                "token_type": "aws_credentials",
             },
         )
         c = Client(api_key="12345", url="https://connect.example/")
         c._ctx.version = None
-        creds = c.oauth.get_content_credentials("cit")
-        assert creds.get("access_token") == "content-token"
+        creds = c.oauth.get_content_credentials("cit", OAuthTokenType.AWS_CREDENTIALS)
+        assert creds.get("access_token") == "encoded-aws-creds"
+        assert creds.get("issued_token_type") == "urn:ietf:params:aws:token-type:credentials"
+        assert creds.get("token_type") == "aws_credentials"
 
     @patch.dict("os.environ", {"CONNECT_CONTENT_SESSION_TOKEN": "cit"})
     @responses.activate