Merge pull request #8 from angelakis/feature-token-exchange

Feature token exchange

Author: rohe

src/oidcmsg/oauth2/__init__.py

@@ -295,6 +295,32 @@ class TokenIntrospectionResponse(Message):
     }
 
 
+# RFC 8693
+class TokenExchangeRequest(Message):
+    c_param = {
+        "grant_type": SINGLE_REQUIRED_STRING,
+        "resource": OPTIONAL_LIST_OF_STRINGS,
+        "audience": OPTIONAL_LIST_OF_STRINGS,
+        "scope": OPTIONAL_LIST_OF_SP_SEP_STRINGS,
+        "requested_token_type": SINGLE_OPTIONAL_STRING,
+        "subject_token": SINGLE_REQUIRED_STRING,
+        "subject_token_type": SINGLE_REQUIRED_STRING,
+        "actor_token": SINGLE_OPTIONAL_STRING,
+        "actor_token_type": SINGLE_OPTIONAL_STRING,
+    }
+
+
+class TokenExchangeResponse(Message):
+    c_param = {
+        "access_token": SINGLE_REQUIRED_STRING,
+        "issued_token_type": SINGLE_REQUIRED_STRING,
+        "token_type": SINGLE_REQUIRED_STRING,
+        "expires_in": SINGLE_OPTIONAL_INT,
+        "refresh_token": SINGLE_OPTIONAL_STRING,
+        "scope": OPTIONAL_LIST_OF_SP_SEP_STRINGS,
+    }
+
+
 class JWTSecuredAuthorizationRequest(AuthorizationRequest):
     c_param = AuthorizationRequest.c_param.copy()
     c_param.update({

tests/test_05_oauth2.py

@@ -18,6 +18,8 @@
 from oidcmsg.oauth2 import AuthorizationResponse
 from oidcmsg.oauth2 import CCAccessTokenRequest
 from oidcmsg.oauth2 import RefreshAccessTokenRequest
+from oidcmsg.oauth2 import TokenExchangeRequest
+from oidcmsg.oauth2 import TokenExchangeResponse
 from oidcmsg.oauth2 import ResponseMessage
 from oidcmsg.oauth2 import ROPCAccessTokenRequest
 from oidcmsg.oauth2 import TokenErrorResponse
@@ -504,6 +506,45 @@ def test_init(self):
         assert ratr.verify()
 
 
+class TestTokenExchangeRequest(object):
+    def test_init(self):
+        request = TokenExchangeRequest(
+            grant_type="urn:ietf:params:oauth:grant-type:token-exchange",
+            subject_token="ababababab",
+            subject_token_type="urn:ietf:params:oauth:token-type:access_token"
+        )
+        assert (
+            request["grant_type"]
+            == "urn:ietf:params:oauth:grant-type:token-exchange"
+        )
+        assert request["subject_token"] == "ababababab"
+        assert (
+            request["subject_token_type"]
+            == "urn:ietf:params:oauth:token-type:access_token"
+        )
+
+        assert request.verify()
+
+
+class TestTokenExchangeResponse(object):
+    def test_init(self):
+        response = TokenExchangeResponse(
+            access_token="bababababa",
+            issued_token_type="urn:ietf:params:oauth:token-type:access_token",
+            token_type="Bearer",
+            expires_in=60
+        )
+        assert (
+            response["issued_token_type"]
+            == "urn:ietf:params:oauth:token-type:access_token"
+        )
+        assert response["access_token"] == "bababababa"
+        assert response["token_type"] == "Bearer"
+        assert response["expires_in"] == 60
+
+        assert response.verify()
+
+
 class TestResponseMessage_error(object):
     def test_error_message(self):
         err = ResponseMessage(error="invalid_request",