Added state parameter to ORCID authorization request

Author: sgomez

src/satosa/backends/orcid.py

@@ -13,7 +13,7 @@
 from satosa.backends.oauth import _OAuthBackend
 from satosa.internal import InternalData
 from satosa.internal import AuthenticationInformation
-from satosa.response import Redirect
+from satosa.util import rndstr
 
 logger = logging.getLogger(__name__)
 
@@ -45,50 +45,41 @@ def __init__(self, outgoing, internal_attributes, config, base_url, name):
             outgoing, internal_attributes, config, base_url, name, 'orcid',
             'orcid')
 
-    def start_auth(self, context, internal_request, get_state=stateID):
-        """
-        :param get_state: Generates a state to be used in authentication call
-
-        :type get_state: Callable[[str, bytes], str]
-        :type context: satosa.context.Context
-        :type internal_request: satosa.internal.InternalData
-        :rtype satosa.response.Redirect
-        """
-        request_args = dict(
-            client_id=self.config['client_config']['client_id'],
-            redirect_uri=self.redirect_url,
-            scope=' '.join(self.config['scope']), )
-        cis = self.consumer.construct_AuthorizationRequest(
-            request_args=request_args)
-        return Redirect(cis.request(self.consumer.authorization_endpoint))
+    def get_request_args(self, get_state=stateID):
+        oauth_state = get_state(self.config["base_url"], rndstr().encode())
+        request_args = {
+            "client_id": self.config['client_config']['client_id'],
+            "redirect_uri": self.redirect_url,
+            "scope": ' '.join(self.config['scope']),
+            "state": oauth_state,
+        }
+        return request_args
 
     def auth_info(self, requrest):
         return AuthenticationInformation(
             UNSPECIFIED, None,
             self.config['server_info']['authorization_endpoint'])
 
     def _authn_response(self, context):
+        state_data = context.state[self.name]
         aresp = self.consumer.parse_response(
             AuthorizationResponse, info=json.dumps(context.request))
-        url = self.config['server_info']['token_endpoint']
-        data = dict(
-            grant_type='authorization_code',
-            code=aresp['code'],
-            redirect_uri=self.redirect_url,
-            client_id=self.config['client_config']['client_id'],
-            client_secret=self.config['client_secret'], )
-        headers = {'Accept': 'application/json'}
+        self._verify_state(aresp, state_data, context.state)
+
+        rargs = {"code": aresp["code"], "redirect_uri": self.redirect_url,
+                 "state": state_data["state"]}
+
+        atresp = self.consumer.do_access_token_request(
+            request_args=rargs, state=aresp['state'])
 
-        r = requests.post(url, data=data, headers=headers)
-        response = r.json()
-        token = response['access_token']
-        orcid, name = response['orcid'], response['name']
-        user_info = self.user_information(token, orcid, name)
-        auth_info = self.auth_info(context.request)
-        internal_response = InternalData(auth_info=auth_info)
+        user_info = self.user_information(
+            atresp['access_token'], atresp['orcid'], atresp['name'])
+        internal_response = InternalData(
+            auth_info=self.auth_info(context.request))
         internal_response.attributes = self.converter.to_internal(
             self.external_type, user_info)
-        internal_response.subject_id = orcid
+        internal_response.subject_id = user_info[self.user_id_attr]
+        del context.state[self.name]
         return self.auth_callback_func(context, internal_response)
 
     def user_information(self, access_token, orcid, name):

tests/satosa/backends/test_orcid.py

@@ -12,10 +12,13 @@
 ORCID_PERSON_ID = "0000-0000-0000-0000"
 ORCID_PERSON_GIVEN_NAME = "orcid_given_name"
 ORCID_PERSON_FAMILY_NAME = "orcid_family_name"
-ORCID_PERSON_NAME = "{} {}".format(ORCID_PERSON_GIVEN_NAME, ORCID_PERSON_FAMILY_NAME)
+ORCID_PERSON_NAME = "{} {}".format(
+    ORCID_PERSON_GIVEN_NAME, ORCID_PERSON_FAMILY_NAME)
 ORCID_PERSON_EMAIL = "orcid_email"
 ORCID_PERSON_COUNTRY = "XX"
 
+mock_get_state = Mock(return_value="abcdef")
+
 
 class TestOrcidBackend(object):
     @pytest.fixture(autouse=True)
@@ -134,7 +137,8 @@ def setup_token_endpoint(self, token_endpoint_url):
     def setup_userinfo_endpoint(self, userinfo_endpoint_url, userinfo):
         responses.add(
             responses.GET,
-            urljoin(userinfo_endpoint_url, '{}/person'.format(ORCID_PERSON_ID)),
+            urljoin(userinfo_endpoint_url,
+                    '{}/person'.format(ORCID_PERSON_ID)),
             body=json.dumps(userinfo),
             status=200,
             content_type="application/json"
@@ -143,19 +147,24 @@ def setup_userinfo_endpoint(self, userinfo_endpoint_url, userinfo):
     @pytest.fixture
     def incoming_authn_response(self, context, backend_config):
         context.path = backend_config["authz_page"]
+        state_data = dict(state=mock_get_state.return_value)
+        context.state[self.orcid_backend.name] = state_data
         context.request = {
             "code": "the_orcid_code",
+            "state": mock_get_state.return_value
         }
 
         return context
 
     def test_start_auth(self, context, backend_config):
-        auth_response = self.orcid_backend.start_auth(context, None)
+        auth_response = self.orcid_backend.start_auth(
+            context, None, mock_get_state)
         assert isinstance(auth_response, Response)
 
         login_url = auth_response.message
         parsed = urlparse(login_url)
-        assert login_url.startswith(backend_config["server_info"]["authorization_endpoint"])
+        assert login_url.startswith(
+            backend_config["server_info"]["authorization_endpoint"])
         auth_params = dict(parse_qsl(parsed.query))
         assert auth_params["scope"] == " ".join(backend_config["scope"])
         assert auth_params["response_type"] == backend_config["response_type"]
@@ -164,11 +173,14 @@ def test_start_auth(self, context, backend_config):
             backend_config["base_url"],
             backend_config["authz_page"]
         )
+        assert auth_params["state"] == mock_get_state.return_value
 
     @responses.activate
     def test_authn_response(self, backend_config, userinfo, incoming_authn_response):
-        self.setup_token_endpoint(backend_config["server_info"]["token_endpoint"])
-        self.setup_userinfo_endpoint(backend_config["server_info"]["user_info"], userinfo)
+        self.setup_token_endpoint(
+            backend_config["server_info"]["token_endpoint"])
+        self.setup_userinfo_endpoint(
+            backend_config["server_info"]["user_info"], userinfo)
 
         self.orcid_backend._authn_response(incoming_authn_response)