text: import json
  import time
  import uuid
  from base64 import b64encode
  
  try:
      from urllib.parse import urlencode
  except ImportError:
      from urllib import urlencode
  
  from django.core.management import call_command
  from django.http import JsonResponse
  
  try:
      from django.urls import reverse
  except ImportError:
      from django.core.urlresolvers import reverse
  
  from django.test import RequestFactory, TestCase, override_settings
  from django.views.decorators.http import require_http_methods
  from jwkest.jwk import KEYS
  from jwkest.jws import JWS
  from jwkest.jwt import JWT
  from mock import patch
  
  from oidc_provider.lib.endpoints.introspection import INTROSPECTION_SCOPE
  from oidc_provider.lib.utils.oauth2 import protected_resource_view
  from oidc_provider.lib.utils.token import create_code
  from oidc_provider.models import Token
  from oidc_provider.tests.app.utils import (
      FAKE_CODE_CHALLENGE,
      FAKE_CODE_VERIFIER,
      FAKE_NONCE,
      FAKE_RANDOM_STRING,
      create_fake_client,
      create_fake_user,
  )
  from oidc_provider.views import JwksView, TokenView, userinfo
  
  
  class TokenTestCase(TestCase):
      """
      To obtain an Access Token and an ID Token, the RP Client sends a
      Token Request to the Token Endpoint to obtain a Token Response
      when using the Authorization Code Flow.
      """
  
      def setUp(self):
          call_command("creatersakey")
          self.factory = RequestFactory()
          self.user = create_fake_user()
          self.request_client = self.client
          self.client = create_fake_client(response_type="code")
  
      def _password_grant_post_data(self, scope=None):
          result = {
              "username": "johndoe",
              "password": "1234",
              "grant_type": "password",
              "scope": "openid email",
          }
          if scope is not None:
              result["scope"] = " ".join(scope)
          return result
  
      def _auth_code_post_data(self, code, scope=None):
          """
          All the data that will be POSTed to the Token Endpoint.
          """
          post_data = {
              "client_id": self.client.client_id,
              "client_secret": self.client.client_secret,
              "redirect_uri": self.client.default_redirect_uri,
              "grant_type": "authorization_code",
              "code": code,
              "state": uuid.uuid4().hex,
          }
          if scope is not None:
              post_data["scope"] = " ".join(scope)
  
          return post_data
  
      def _refresh_token_post_data(self, refresh_token, scope=None):
          """
          All the data that will be POSTed to the Token Endpoint.
          """
          post_data = {
              "client_id": self.client.client_id,
              "client_secret": self.client.client_secret,
              "grant_type": "refresh_token",
              "refresh_token": refresh_token,
          }
          if scope is not None:
              post_data["scope"] = " ".join(scope)
  
          return post_data
  
      def _post_request(self, post_data, extras={}):
          """
          Makes a request to the token endpoint by sending the
          `post_data` parameters using the 'application/x-www-form-urlencoded'
          format.
          """
          url = reverse("oidc_provider:token")
  
          request = self.factory.post(
              url,
              data=urlencode(post_data),
              content_type="application/x-www-form-urlencoded",
              **extras
          )
  
          response = TokenView.as_view()(request)
  
          return response
  
      def _create_code(self, scope=None):
          """
          Generate a valid grant code.
          """
          code = create_code(
              user=self.user,
              client=self.client,
              scope=(scope if scope else ["openid", "email"]),
              nonce=FAKE_NONCE,
              is_authentication=True,
          )
          code.save()
  
          return code
  
      def _get_keys(self):
          """
          Get public key from discovery.
          """
          request = self.factory.get(reverse("oidc_provider:jwks"))
          response = JwksView.as_view()(request)
          jwks_dic = json.loads(response.content.decode("utf-8"))
          SIGKEYS = KEYS()
          SIGKEYS.load_dict(jwks_dic)
          return SIGKEYS
  
      def _get_userinfo(self, access_token):
          url = reverse("oidc_provider:userinfo")
          request = self.factory.get(url)
          request.META["HTTP_AUTHORIZATION"] = "Bearer " + access_token
  
          return userinfo(request)
  
      def _password_grant_auth_header(self):
          user_pass = self.client.client_id + ":" + self.client.client_secret
          auth = b"Basic " + b64encode(user_pass.encode("utf-8"))
          auth_header = {"HTTP_AUTHORIZATION": auth.decode("utf-8")}
          return auth_header
  
      def test_default_setting_does_not_allow_grant_type_password(self):
          post_data = self._password_grant_post_data()
  
          response = self._post_request(
              post_data=post_data, extras=self._password_grant_auth_header()
          )
  
          response_dict = json.loads(response.content.decode("utf-8"))
  
          self.assertEqual(400, response.status_code)
          self.assertEqual("unsupported_grant_type", response_dict["error"])
  
      @override_settings(OIDC_GRANT_TYPE_PASSWORD_ENABLE=True)
      def test_password_grant_get_access_token_without_scope(self):
          post_data = self._password_grant_post_data()
          del post_data["scope"]
  
          response = self._post_request(
              post_data=post_data, extras=self._password_grant_auth_header()
          )
  
          response_dict = json.loads(response.content.decode("utf-8"))
          self.assertIn("access_token", response_dict)
  
      @override_settings(OIDC_GRANT_TYPE_PASSWORD_ENABLE=True)
      def test_password_grant_get_access_token_with_scope(self):
          response = self._post_request(
              post_data=self._password_grant_post_data(),
              extras=self._password_grant_auth_header(),
          )
  
          response_dict = json.loads(response.content.decode("utf-8"))
          self.assertIn("access_token", response_dict)
  
      @override_settings(OIDC_GRANT_TYPE_PASSWORD_ENABLE=True)
      def test_password_grant_get_access_token_invalid_user_credentials(self):
          invalid_post = self._password_grant_post_data()
          invalid_post["password"] = "wrong!"
  
          response = self._post_request(
              post_data=invalid_post, extras=self._password_grant_auth_header()
          )
  
          response_dict = json.loads(response.content.decode("utf-8"))
  
          self.assertEqual(403, response.status_code)
          self.assertEqual("access_denied", response_dict["error"])
  
      def test_password_grant_get_access_token_invalid_client_credentials(self):
          self.client.client_id = "foo"
          self.client.client_secret = "bar"
  
          response = self._post_request(
              post_data=self._password_grant_post_data(),
              extras=self._password_grant_auth_header(),
          )
  
          response_dict = json.loads(response.content.decode("utf-8"))
  
          self.assertEqual(400, response.status_code)
          self.assertEqual("invalid_client", response_dict["error"])
  
      def test_password_grant_full_response(self):
          self.check_password_grant(scope=["openid", "email"])
  
      def test_password_grant_scope(self):
          self.check_password_grant(scope=["openid", "profile"])
  
      @override_settings(OIDC_TOKEN_EXPIRE=120, OIDC_GRANT_TYPE_PASSWORD_ENABLE=True)
      def check_password_grant(self, scope):
          response = self._post_request(
              post_data=self._password_grant_post_data(scope),
              extras=self._password_grant_auth_header(),
          )
  
          response_dict = json.loads(response.content.decode("utf-8"))
          id_token = JWS().verify_compact(
              response_dict["id_token"].encode("utf-8"), self._get_keys()
          )
  
          token = Token.objects.get(user=self.user)
          self.assertEqual(response_dict["access_token"], token.access_token)
          self.assertEqual(response_dict["refresh_token"], token.refresh_token)
          self.assertEqual(response_dict["expires_in"], 120)
          self.assertEqual(response_dict["token_type"], "bearer")
          self.assertEqual(id_token["sub"], str(self.user.id))
          self.assertEqual(id_token["aud"], self.client.client_id)
  
          # Check the scope is honored by checking the claims in the userinfo
          userinfo_response = self._get_userinfo(response_dict["access_token"])
          userinfo = json.loads(userinfo_response.content.decode("utf-8"))
  
          for (scope_param, claim) in [("email", "email"), ("profile", "name")]:
              if scope_param in scope:
                  self.assertIn(claim, userinfo)
              else:
                  self.assertNotIn(claim, userinfo)
  
      @override_settings(
          OIDC_GRANT_TYPE_PASSWORD_ENABLE=True,
          AUTHENTICATION_BACKENDS=("oidc_provider.tests.app.utils.TestAuthBackend",),
      )
      def test_password_grant_passes_request_to_backend(self):
          response = self._post_request(
              post_data=self._password_grant_post_data(),
              extras=self._password_grant_auth_header(),
          )
  
          response_dict = json.loads(response.content.decode("utf-8"))
          self.assertIn("access_token", response_dict)
  
      @override_settings(OIDC_TOKEN_EXPIRE=720)
      def test_authorization_code(self):
          """
          We MUST validate the signature of the ID Token according to JWS
          using the algorithm specified in the alg Header Parameter of
          the JOSE Header.
          """
          SIGKEYS = self._get_keys()
          code = self._create_code()
  
          post_data = self._auth_code_post_data(code=code.code)
  
          response = self._post_request(post_data)
          response_dic = json.loads(response.content.decode("utf-8"))
  
          id_token = JWS().verify_compact(
              response_dic["id_token"].encode("utf-8"), SIGKEYS
          )
  
          token = Token.objects.get(user=self.user)
          self.assertEqual(response_dic["access_token"], token.access_token)
          self.assertEqual(response_dic["refresh_token"], token.refresh_token)
          self.assertEqual(response_dic["token_type"], "bearer")
          self.assertEqual(response_dic["expires_in"], 720)
          self.assertEqual(id_token["sub"], str(self.user.id))
          self.assertEqual(id_token["aud"], self.client.client_id)
  
      @override_settings(OIDC_TOKEN_EXPIRE=720, OIDC_IDTOKEN_INCLUDE_CLAIMS=True)
      def test_scope_is_ignored_for_auth_code(self):
          """
          Scope is ignored for token respones to auth code grant type.
          This comes down to that the scopes requested in authorize are returned.
          """
          SIGKEYS = self._get_keys()
          for code_scope in [["openid"], ["openid", "email"], ["openid", "profile"]]:
              code = self._create_code(code_scope)
  
              post_data = self._auth_code_post_data(code=code.code, scope=code_scope)
  
              response = self._post_request(post_data)
              response_dic = json.loads(response.content.decode("utf-8"))
  
              self.assertEqual(response.status_code, 200)
  
              id_token = JWS().verify_compact(
                  response_dic["id_token"].encode("utf-8"), SIGKEYS
              )
  
              if "email" in code_scope:
                  self.assertIn("email", id_token)
                  self.assertIn("email_verified", id_token)
              else:
                  self.assertNotIn("email", id_token)
  
              if "profile" in code_scope:
                  self.assertIn("given_name", id_token)
              else:
                  self.assertNotIn("given_name", id_token)
  
      def test_refresh_token(self):
          """
          A request to the Token Endpoint can also use a Refresh Token
          by using the grant_type value refresh_token, as described in
          Section 6 of OAuth 2.0 [RFC6749].
          """
          self.do_refresh_token_check()
  
      def test_refresh_token_invalid_scope(self):
          """
          Extending scope in refresh token is not allowed.
  
          Try to get a refresh token with "profile" in the scope even
          though the original authorized scope in the authorization code
          request is only ['openid', 'email'].
          """
          self.do_refresh_token_check(scope=["openid", "profile"])
  
      def test_refresh_token_narrowed_scope(self):
          """
          Narrowing scope in refresh token is allowed.
  
          Try to get a refresh token with just "openid" in the scope even
          though the original authorized scope in the authorization code
          request is ['openid', 'email'].
          """
          self.do_refresh_token_check(scope=["openid"])
  
      @override_settings(OIDC_IDTOKEN_INCLUDE_CLAIMS=True)
      def do_refresh_token_check(self, scope=None):
          SIGKEYS = self._get_keys()
  
          # Retrieve refresh token
          code = self._create_code()
          self.assertEqual(code.scope, ["openid", "email"])
          post_data = self._auth_code_post_data(code=code.code)
          start_time = time.time()
          with patch("oidc_provider.lib.utils.token.time.time") as time_func:
              time_func.return_value = start_time
              response = self._post_request(post_data)
  
          response_dic1 = json.loads(response.content.decode("utf-8"))
          id_token1 = JWS().verify_compact(
              response_dic1["id_token"].encode("utf-8"), SIGKEYS
          )
  
          # Use refresh token to obtain new token
          post_data = self._refresh_token_post_data(response_dic1["refresh_token"], scope)
          with patch("oidc_provider.lib.utils.token.time.time") as time_func:
              time_func.return_value = start_time + 600
              response = self._post_request(post_data)
  
          response_dic2 = json.loads(response.content.decode("utf-8"))
  
          if scope and set(scope) - set(code.scope):  # too broad scope
              self.assertEqual(response.status_code, 400)  # Bad Request
              self.assertIn("error", response_dic2)
              self.assertEqual(response_dic2["error"], "invalid_scope")
              return  # No more checks
  
          id_token2 = JWS().verify_compact(
              response_dic2["id_token"].encode("utf-8"), SIGKEYS
          )
  
          if scope and "email" not in scope:  # narrowed scope The auth
              # The auth code request had email in scope, so it should be
              # in the first id token
              self.assertIn("email", id_token1)
              # but the refresh request had no email in scope
              self.assertNotIn("email", id_token2, "email was not requested")
  
          self.assertNotEqual(response_dic1["id_token"], response_dic2["id_token"])
          self.assertNotEqual(
              response_dic1["access_token"], response_dic2["access_token"]
          )
          self.assertNotEqual(
              response_dic1["refresh_token"], response_dic2["refresh_token"]
          )
  
          # http://openid.net/specs/openid-connect-core-1_0.html#rfc.section.12.2
          self.assertEqual(id_token1["iss"], id_token2["iss"])
          self.assertEqual(id_token1["sub"], id_token2["sub"])
          self.assertNotEqual(id_token1["iat"], id_token2["iat"])
          self.assertEqual(id_token1["iat"], int(start_time))
          self.assertEqual(id_token2["iat"], int(start_time + 600))
          self.assertEqual(id_token1["aud"], id_token2["aud"])
          self.assertEqual(id_token1["auth_time"], id_token2["auth_time"])
          self.assertEqual(id_token1.get("azp"), id_token2.get("azp"))
  
          # Refresh token can't be reused
          post_data = self._refresh_token_post_data(response_dic1["refresh_token"])
          response = self._post_request(post_data)
          self.assertIn("invalid_grant", response.content.decode("utf-8"))
  
          # Old access token is invalidated
          self.assertEqual(
              self._get_userinfo(response_dic1["access_token"]).status_code, 401
          )
          self.assertEqual(
              self._get_userinfo(response_dic2["access_token"]).status_code, 200
          )
  
          # Empty refresh token is invalid
          post_data = self._refresh_token_post_data("")
          response = self._post_request(post_data)
          self.assertIn("invalid_grant", response.content.decode("utf-8"))
  
          # No refresh token is invalid
          post_data = self._refresh_token_post_data("")
          del post_data["refresh_token"]
          response = self._post_request(post_data)
          self.assertIn("invalid_grant", response.content.decode("utf-8"))
  
      def test_client_redirect_uri(self):
          """
          Validate that client redirect URIs exactly match registered
          URIs, and that unregistered URIs or URIs with query parameters are rejected.
          See http://openid.net/specs/openid-connect-core-1_0.html#AuthRequest and
          http://openid.net/specs/openid-connect-core-1_0.html#HybridTokenRequest.
          """
          code = self._create_code()
          post_data = self._auth_code_post_data(code=code.code)
  
          # Unregistered URI
          post_data["redirect_uri"] = "http://invalid.example.org"
  
          response = self._post_request(post_data)
          self.assertIn("invalid_client", response.content.decode("utf-8"))
  
          # Registered URI, but with query string appended
          post_data["redirect_uri"] = self.client.default_redirect_uri + "?foo=bar"
  
          response = self._post_request(post_data)
          self.assertIn("invalid_client", response.content.decode("utf-8"))
  
          # Registered URI
          post_data["redirect_uri"] = self.client.default_redirect_uri
  
          response = self._post_request(post_data)
          self.assertNotIn("invalid_client", response.content.decode("utf-8"))
  
      def test_request_methods(self):
          """
          Client sends an HTTP POST request to the Token Endpoint. Other request
          methods MUST NOT be allowed.
          """
          url = reverse("oidc_provider:token")
  
          requests = [
              self.factory.get(url),
              self.factory.put(url),
              self.factory.delete(url),
          ]
  
          for request in requests:
              response = TokenView.as_view()(request)
  
              self.assertEqual(
                  response.status_code,
                  405,
                  msg=request.method + " request does not return a 405 status.",
              )
  
          request = self.factory.post(url)
  
          response = TokenView.as_view()(request)
  
          self.assertEqual(
              response.status_code,
              400,
              msg=request.method + " request does not return a 400 status.",
          )
  
      def test_client_authentication(self):
          """
          The authorization server support including the
          client credentials in the request-body using the `client_id` and
          `client_secret`parameters.
  
          See: http://tools.ietf.org/html/rfc6749#section-2.3.1
          """
          code = self._create_code()
  
          # Test a valid request to the token endpoint.
          post_data = self._auth_code_post_data(code=code.code)
  
          response = self._post_request(post_data)
  
          self.assertNotIn(
              "invalid_client",
              response.content.decode("utf-8"),
              msg="Client authentication fails using request-body credentials.",
          )
  
          # Now, test with an invalid client_id.
          invalid_data = post_data.copy()
          invalid_data["client_id"] = self.client.client_id * 2  # Fake id.
  
          # Create another grant code.
          code = self._create_code()
          invalid_data["code"] = code.code
  
          response = self._post_request(invalid_data)
  
          self.assertIn(
              "invalid_client",
              response.content.decode("utf-8"),
              msg='Client authentication success with an invalid "client_id".',
          )
  
          # Now, test using HTTP Basic Authentication method.
          basicauth_data = post_data.copy()
  
          # Create another grant code.
          code = self._create_code()
          basicauth_data["code"] = code.code
  
          del basicauth_data["client_id"]
          del basicauth_data["client_secret"]
  
          response = self._post_request(
              basicauth_data, self._password_grant_auth_header()
          )
          response.content.decode("utf-8")
  
          self.assertNotIn(
              "invalid_client",
              response.content.decode("utf-8"),
              msg="Client authentication fails using HTTP Basic Auth.",
          )
  
      def test_access_token_contains_nonce(self):
          """
          If present in the Authentication Request, Authorization Servers MUST
          include a nonce Claim in the ID Token with the Claim Value being the
          nonce value sent in the Authentication Request.
          If the client does not supply a nonce parameter, it SHOULD not be
          included in the `id_token`.
  
          See http://openid.net/specs/openid-connect-core-1_0.html#IDToken
          """
          code = self._create_code()
  
          post_data = self._auth_code_post_data(code=code.code)
  
          response = self._post_request(post_data)
  
          response_dic = json.loads(response.content.decode("utf-8"))
          id_token = JWT().unpack(response_dic["id_token"].encode("utf-8")).payload()
  
          self.assertEqual(id_token.get("nonce"), FAKE_NONCE)
  
          # Client does not supply a nonce parameter.
          code.nonce = ""
          code.save()
  
          response = self._post_request(post_data)
          response_dic = json.loads(response.content.decode("utf-8"))
  
          id_token = JWT().unpack(response_dic["id_token"].encode("utf-8")).payload()
  
          self.assertEqual(id_token.get("nonce"), None)
  
      def test_id_token_contains_at_hash(self):
          """
          If access_token is included, the id_token SHOULD contain an at_hash.
          """
          code = self._create_code()
  
          post_data = self._auth_code_post_data(code=code.code)
  
          response = self._post_request(post_data)
  
          response_dic = json.loads(response.content.decode("utf-8"))
          id_token = JWT().unpack(response_dic["id_token"].encode("utf-8")).payload()
  
          self.assertTrue(id_token.get("at_hash"))
  
      def test_idtoken_sign_validation(self):
          """
          We MUST validate the signature of the ID Token according to JWS
          using the algorithm specified in the alg Header Parameter of
          the JOSE Header.
          """
          SIGKEYS = self._get_keys()
          RSAKEYS = [k for k in SIGKEYS if k.kty == "RSA"]
  
          code = self._create_code()
  
          post_data = self._auth_code_post_data(code=code.code)
  
          response = self._post_request(post_data)
          response_dic = json.loads(response.content.decode("utf-8"))
  
          JWS().verify_compact(response_dic["id_token"].encode("utf-8"), RSAKEYS)
  
      @override_settings(
          OIDC_IDTOKEN_SUB_GENERATOR="oidc_provider.tests.app.utils.fake_sub_generator"
      )
      def test_custom_sub_generator(self):
          """
          Test custom function for setting OIDC_IDTOKEN_SUB_GENERATOR.
          """
          code = self._create_code()
  
          post_data = self._auth_code_post_data(code=code.code)
  
          response = self._post_request(post_data)
  
          response_dic = json.loads(response.content.decode("utf-8"))
          id_token = JWT().unpack(response_dic["id_token"].encode("utf-8")).payload()
  
          self.assertEqual(id_token.get("sub"), self.user.email)
  
      @override_settings(
          OIDC_IDTOKEN_PROCESSING_HOOK="oidc_provider.tests.app.utils.fake_idtoken_processing_hook"
      )
      def test_additional_idtoken_processing_hook(self):
          """
          Test custom function for setting OIDC_IDTOKEN_PROCESSING_HOOK.
          """
          code = self._create_code()
  
          post_data = self._auth_code_post_data(code=code.code)
  
          response = self._post_request(post_data)
  
          response_dic = json.loads(response.content.decode("utf-8"))
          id_token = JWT().unpack(response_dic["id_token"].encode("utf-8")).payload()
  
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook"), FAKE_RANDOM_STRING
          )
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook_user_email"), self.user.email
          )
  
      @override_settings(
          OIDC_IDTOKEN_PROCESSING_HOOK=(
              "oidc_provider.tests.app.utils.fake_idtoken_processing_hook",
          )
      )
      def test_additional_idtoken_processing_hook_one_element_in_tuple(self):
          """
          Test custom function for setting OIDC_IDTOKEN_PROCESSING_HOOK.
          """
          code = self._create_code()
  
          post_data = self._auth_code_post_data(code=code.code)
  
          response = self._post_request(post_data)
  
          response_dic = json.loads(response.content.decode("utf-8"))
          id_token = JWT().unpack(response_dic["id_token"].encode("utf-8")).payload()
  
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook"), FAKE_RANDOM_STRING
          )
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook_user_email"), self.user.email
          )
  
      @override_settings(
          OIDC_IDTOKEN_PROCESSING_HOOK=[
              "oidc_provider.tests.app.utils.fake_idtoken_processing_hook",
          ]
      )
      def test_additional_idtoken_processing_hook_one_element_in_list(self):
          """
          Test custom function for setting OIDC_IDTOKEN_PROCESSING_HOOK.
          """
          code = self._create_code()
  
          post_data = self._auth_code_post_data(code=code.code)
  
          response = self._post_request(post_data)
  
          response_dic = json.loads(response.content.decode("utf-8"))
          id_token = JWT().unpack(response_dic["id_token"].encode("utf-8")).payload()
  
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook"), FAKE_RANDOM_STRING
          )
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook_user_email"), self.user.email
          )
  
      @override_settings(
          OIDC_IDTOKEN_PROCESSING_HOOK=[
              "oidc_provider.tests.app.utils.fake_idtoken_processing_hook",
              "oidc_provider.tests.app.utils.fake_idtoken_processing_hook2",
          ]
      )
      def test_additional_idtoken_processing_hook_two_elements_in_list(self):
          """
          Test custom function for setting OIDC_IDTOKEN_PROCESSING_HOOK.
          """
          code = self._create_code()
  
          post_data = self._auth_code_post_data(code=code.code)
  
          response = self._post_request(post_data)
  
          response_dic = json.loads(response.content.decode("utf-8"))
          id_token = JWT().unpack(response_dic["id_token"].encode("utf-8")).payload()
  
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook"), FAKE_RANDOM_STRING
          )
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook_user_email"), self.user.email
          )
  
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook2"), FAKE_RANDOM_STRING
          )
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook_user_email2"), self.user.email
          )
  
      @override_settings(
          OIDC_IDTOKEN_PROCESSING_HOOK=(
              "oidc_provider.tests.app.utils.fake_idtoken_processing_hook",
              "oidc_provider.tests.app.utils.fake_idtoken_processing_hook2",
          )
      )
      def test_additional_idtoken_processing_hook_two_elements_in_tuple(self):
          """
          Test custom function for setting OIDC_IDTOKEN_PROCESSING_HOOK.
          """
          code = self._create_code()
  
          post_data = self._auth_code_post_data(code=code.code)
  
          response = self._post_request(post_data)
  
          response_dic = json.loads(response.content.decode("utf-8"))
          id_token = JWT().unpack(response_dic["id_token"].encode("utf-8")).payload()
  
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook"), FAKE_RANDOM_STRING
          )
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook_user_email"), self.user.email
          )
  
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook2"), FAKE_RANDOM_STRING
          )
          self.assertEqual(
              id_token.get("test_idtoken_processing_hook_user_email2"), self.user.email
          )
  
      @override_settings(
          OIDC_IDTOKEN_PROCESSING_HOOK=(
              "oidc_provider.tests.app.utils.fake_idtoken_processing_hook3"
          )
      )
      def test_additional_idtoken_processing_hook_scope_available(self):
          """
          Test scope is available in OIDC_IDTOKEN_PROCESSING_HOOK.
          """
          id_token = self._request_id_token_with_scope(
              ["openid", "email", "profile", "dummy"]
          )
          self.assertEqual(
              id_token.get("scope_of_token_passed_to_processing_hook"),
              ["openid", "email", "profile", "dummy"],
          )
  
      @override_settings(
          OIDC_IDTOKEN_PROCESSING_HOOK=(
              "oidc_provider.tests.app.utils.fake_idtoken_processing_hook4"
          )
      )
      def test_additional_idtoken_processing_hook_kwargs(self):
          """
          Test correct kwargs are passed to OIDC_IDTOKEN_PROCESSING_HOOK.
          """
          id_token = self._request_id_token_with_scope(["openid", "profile"])
          kwargs_passed = id_token.get("kwargs_passed_to_processing_hook")
          assert kwargs_passed
          self.assertTrue(kwargs_passed.get("token").startswith("<Token: Some Client -"))
          self.assertEqual(
              kwargs_passed.get("request"), "<WSGIRequest: POST '/openid/token'>"
          )
          self.assertEqual(set(kwargs_passed.keys()), {"token", "request"})
  
      def _request_id_token_with_scope(self, scope):
          code = self._create_code(scope)
  
          post_data = self._auth_code_post_data(code=code.code)
  
          response = self._post_request(post_data)
  
          response_dic = json.loads(response.content.decode("utf-8"))
          id_token = JWT().unpack(response_dic["id_token"].encode("utf-8")).payload()
          return id_token
  
      def test_pkce_parameters(self):
          """
          Test Proof Key for Code Exchange by OAuth Public Clients.
          https://tools.ietf.org/html/rfc7636
          """
          code = create_code(
              user=self.user,
              client=self.client,
              scope=["openid", "email"],
              nonce=FAKE_NONCE,
              is_authentication=True,
              code_challenge=FAKE_CODE_CHALLENGE,
              code_challenge_method="S256",
          )
          code.save()
  
          post_data = self._auth_code_post_data(code=code.code)
  
          # Add parameters.
          post_data["code_verifier"] = FAKE_CODE_VERIFIER
  
          response = self._post_request(post_data)
  
          json.loads(response.content.decode("utf-8"))
  
      @override_settings(OIDC_INTROSPECTION_VALIDATE_AUDIENCE_SCOPE=False)
      def test_client_credentials_grant_type(self):
          fake_scopes_list = ["scopeone", "scopetwo", INTROSPECTION_SCOPE]
  
          # Add scope for this client.
          self.client.scope = fake_scopes_list
          self.client.save()
  
          post_data = {
              "client_id": self.client.client_id,
              "client_secret": self.client.client_secret,
              "grant_type": "client_credentials",
          }
          response = self._post_request(post_data)
          response_dict = json.loads(response.content.decode("utf-8"))
  
          # Ensure access token exists in the response, also check if scopes are
          # the ones we registered previously.
          self.assertTrue("access_token" in response_dict)
          self.assertEqual(" ".join(fake_scopes_list), response_dict["scope"])
  
          access_token = response_dict["access_token"]
  
          # Create a protected resource and test the access_token.
  
          @require_http_methods(["GET"])
          @protected_resource_view(fake_scopes_list)
          def protected_api(request, *args, **kwargs):
              return JsonResponse({"protected": "information"}, status=200)
  
          # Deploy view on some url. So, base url could be anything.
          request = self.factory.get(
              "/api/protected/?access_token={0}".format(access_token)
          )
          response = protected_api(request)
          response_dict = json.loads(response.content.decode("utf-8"))
  
          self.assertEqual(response.status_code, 200)
          self.assertTrue("protected" in response_dict)
  
          # Protected resource test ends here.
  
          # Verify access_token can be validated with token introspection
  
          response = self.request_client.post(
              reverse("oidc_provider:token-introspection"),
              data={"token": access_token},
              **self._password_grant_auth_header()
          )
          self.assertEqual(response.status_code, 200)
          response_dict = json.loads(response.content.decode("utf-8"))
          self.assertTrue(response_dict.get("active"))
  
          # End token introspection test
  
          # Clean scopes for this client.
          self.client.scope = ""
          self.client.save()
  
          response = self._post_request(post_data)
          response_dict = json.loads(response.content.decode("utf-8"))
  
          # It should fail when client does not have any scope added.
          self.assertEqual(400, response.status_code)
          self.assertEqual("invalid_scope", response_dict["error"])
  
      def test_printing_token_used_by_client_credentials_grant_type(self):
          # Add scope for this client.
          self.client.scope = ["something"]
          self.client.save()
  
          post_data = {
              "client_id": self.client.client_id,
              "client_secret": self.client.client_secret,
              "grant_type": "client_credentials",
          }
          response = self._post_request(post_data)
          response_dict = json.loads(response.content.decode("utf-8"))
          token = Token.objects.get(access_token=response_dict["access_token"])
          self.assertTrue(str(token))
  