test_token_response.py

import requests
import json
import six
import time
import jwt
try:
    import mock
except ImportError:
    from unittest import mock

from tests.framework import (CapturedIOTestCase, SDKTESTER1A_NATIVE1_ID_TOKEN,
                             SDKTESTER1A_ID_ACCESS_TOKEN, get_client_data,
                             retry_errors)
from globus_sdk.auth.token_response import (
    OAuthTokenResponse, OAuthDependentTokenResponse, _convert_token_info_dict)


class OAuthTokenResponseTests(CapturedIOTestCase):

    def setUp(self):
        """
        Creates a OAuthTokenResponse object with known data for testing
        Sets up mock AuthClient for decoding id_tokens
        """
        super(OAuthTokenResponseTests, self).setUp()
        # known token data for testing expected values
        self.top_token = {  # valid id_token and access_token
            "resource_server": "server1", "expires_in": 10, "scope": "scope1",
            "refresh_token": "RT1", "other_tokens": [], "token_type": "1",
            "id_token": SDKTESTER1A_NATIVE1_ID_TOKEN,
            "access_token": SDKTESTER1A_ID_ACCESS_TOKEN}
        self.other_token1 = {  # invalid id_token with valid access_token
            "resource_server": "server2", "expires_in": 20, "scope": "scope2",
            "refresh_token": "RT2", "other_tokens": [], "token_type": "2",
            "id_token": "invalid_id_token",
            "access_token": SDKTESTER1A_ID_ACCESS_TOKEN}
        self.other_token2 = {  # valid id_token with invalid access_token
            "resource_server": "server3", "expires_in": 30,
            "scope": "scope3 scope4",
            "refresh_token": "RT3", "other_tokens": [], "token_type": "3",
            "id_token": SDKTESTER1A_NATIVE1_ID_TOKEN,
            "access_token": "invalid_access_token"}
        self.top_token["other_tokens"] = [self.other_token1, self.other_token2]

        # mock AuthClient
        self.ac = mock.Mock()
        self.ac.client_id = get_client_data()["native_app_client1"]["id"]
        self.ac._verify = True
        self.ac.get = mock.Mock(return_value={
            "jwks_uri": "https://auth.globus.org/jwk.json",
            "id_token_signing_alg_values_supported": ["RS512"]
        })

        # create the response
        http_response = requests.Response()
        http_response._content = six.b(json.dumps(self.top_token))
        http_response.headers["Content-Type"] = "application/json"
        self.response = OAuthTokenResponse(http_response, client=self.ac)

    def test_convert_token_info_dict(self):
        """
        Converts known info dicts to confirm expected results
        Confirms only refresh_token and token_type may be missing,
        And expires_at is created correctly.
        """
        top_convert = _convert_token_info_dict(self.top_token)

        # expected results for top token data
        for key in ["scope", "access_token", "refresh_token", "token_type"]:
            self.assertEqual(self.top_token[key], top_convert[key])

        # sanity check expires at, assuming test runs within 1 second range
        expected = int(time.time()) + self.top_token["expires_in"]
        self.assertIn(top_convert["expires_at_seconds"],
                      (expected - 1, expected, expected + 1))

        # missing refresh_token or token_type
        for key in ["refresh_token", "token_type"]:
            value = self.top_token.pop(key)
            top_convert = _convert_token_info_dict(self.top_token)
            self.assertIsNone(top_convert[key])
            self.top_token[key] = value

        # required keys
        for key in ["scope", "access_token"]:
            value = self.top_token.pop(key)
            with self.assertRaises(KeyError):
                _convert_token_info_dict(self.top_token)
            self.top_token[key] = value

        # no expires_in makes expires at now, withing a 1 second range
        self.top_token.pop("expires_in")
        top_convert = _convert_token_info_dict(self.top_token)
        expected = int(time.time())
        self.assertIn(top_convert["expires_at_seconds"],
                      (expected - 1, expected, expected + 1))

    def test_by_resource_server(self):
        """
        Gets by_resource_server attribute from test response,
        Confirms expected values found for top and other tokens
        """
        by_server = self.response.by_resource_server

        # confirm data by server matches known token values
        for server, token in [("server1", self.top_token),
                              ("server2", self.other_token1),
                              ("server3", self.other_token2)]:
            server_data = by_server[server]
            for key in ["scope", "access_token",
                        "refresh_token", "token_type"]:
                self.assertEqual(server_data[key], token[key])
            # assumes test runs within 1 second range
            expected = int(time.time()) + token["expires_in"]
            self.assertIn(server_data["expires_at_seconds"],
                          (expected - 1, expected, expected + 1))

    def test_by_scopes(self):
        """
        Gets by_scopes attribute from test response,
        Confirms expected values found for top and other tokens
        """
        by_scopes = self.response.by_scopes

        # confirm data by server matches known token values
        for scope, token in [("scope1", self.top_token),
                             ("scope2", self.other_token1),
                             ("scope3", self.other_token2),
                             ("scope4", self.other_token2),
                             ("scope3 scope4", self.other_token2),
                             ("scope4 scope3", self.other_token2)]:
            scope_data = by_scopes[scope]
            for key in ["scope", "access_token",
                        "refresh_token", "token_type"]:
                self.assertEqual(scope_data[key], token[key])
            # assumes test runs within 1 second range
            expected = int(time.time()) + token["expires_in"]
            self.assertIn(scope_data["expires_at_seconds"],
                          (expected - 1, expected, expected + 1))

        self.assertIn('scope1', by_scopes)
        self.assertIn('scope3', by_scopes)
        self.assertNotIn('scope1 scope2', by_scopes)
        self.assertNotIn('scope1 scope3', by_scopes)
        self.assertIn('scope4 scope3', by_scopes)

    @retry_errors()
    def test_decode_id_token_invalid_id(self):
        """
        Creates a response with an invalid id_token, and attempts to decode
        Confirms JWTError
        """
        http_response = requests.Response()
        http_response._content = six.b(json.dumps(self.other_token1))
        http_response.headers["Content-Type"] = "application/json"
        id_response = OAuthTokenResponse(http_response, client=self.ac)

        with self.assertRaises(jwt.exceptions.InvalidTokenError):
            id_response.decode_id_token()
        # test with deprecated usage pattern too
        with self.assertRaises(jwt.exceptions.InvalidTokenError):
            id_response.decode_id_token(self.ac)

    @retry_errors()
    def test_decode_id_token_expired(self):
        """
        Attempt to decode an expired id_token, confirms that the token is
        decoded, but errors out on the expired signature.
        """
        with self.assertRaises(jwt.exceptions.ExpiredSignatureError):
            self.response.decode_id_token()
        # test with deprecated usage pattern too
        with self.assertRaises(jwt.exceptions.ExpiredSignatureError):
            self.response.decode_id_token(self.ac)


class OAuthDependentTokenResponseTests(CapturedIOTestCase):

    def setUp(self):
        """
        Creates a OAuthDependentTokenResponse with known data for testing
        """
        super(OAuthDependentTokenResponseTests, self).setUp()
        # known token data for testing expected values
        self.token1 = {
            "resource_server": "server1", "expires_in": 10, "scope": "scope1",
            "access_token": "AT1", "refresh_token": "RT1", "token_type": "1"}
        self.token2 = {
            "resource_server": "server2", "expires_in": 20, "scope": "scope2",
            "access_token": "AT2", "refresh_token": "RT2", "token_type": "2"}
        self.token3 = {
            "resource_server": "server3", "expires_in": 30,
            "scope": "scope3 scope4",
            "access_token": "AT3", "refresh_token": "RT3", "token_type": "3"}

        # create the response
        http_response = requests.Response()
        data = [self.token1, self.token2, self.token3]
        http_response._content = six.b(json.dumps(data))
        http_response.headers["Content-Type"] = "application/json"
        self.response = OAuthDependentTokenResponse(http_response)

    def test_by_resource_server(self):
        """
        Gets by_resource_server attribute from test response,
        Confirms expected values found for top and other tokens
        """
        by_server = self.response.by_resource_server

        # confirm data by server matches known token values
        for server, token in [("server1", self.token1),
                              ("server2", self.token2),
                              ("server3", self.token3)]:
            server_data = by_server[server]
            for key in ["scope", "access_token",
                        "refresh_token", "token_type"]:
                self.assertEqual(server_data[key], token[key])
            # assumes test runs within 1 second range
            expected = int(time.time()) + token["expires_in"]
            self.assertIn(server_data["expires_at_seconds"],
                          (expected - 1, expected, expected + 1))

    def test_by_scopes(self):
        """
        Gets by_scopes attribute from test response,
        Confirms expected values found for top and other tokens
        """
        by_scopes = self.response.by_scopes

        # confirm data by server matches known token values
        for scope, token in [("scope1", self.token1),
                             ("scope2", self.token2),
                             ("scope3", self.token3),
                             ("scope4", self.token3),
                             ("scope3 scope4", self.token3),
                             ("scope4 scope3", self.token3)]:
            scope_data = by_scopes[scope]
            for key in ["scope", "access_token",
                        "refresh_token", "token_type"]:
                self.assertEqual(scope_data[key], token[key])
            # assumes test runs within 1 second range
            expected = int(time.time()) + token["expires_in"]
            self.assertIn(scope_data["expires_at_seconds"],
                          (expected - 1, expected, expected + 1))

        self.assertIn('scope1', by_scopes)
        self.assertIn('scope3', by_scopes)
        self.assertNotIn('scope1 scope2', by_scopes)
        self.assertNotIn('scope1 scope3', by_scopes)
        self.assertIn('scope4 scope3', by_scopes)