Added tests

Author: rohe

src/satosa/backends/idpy_oidc.py

@@ -1,17 +1,17 @@
 """
 OIDC/OAuth2 backend module.
 """
-import logging
 from datetime import datetime
+import logging
 from urllib.parse import urlparse
 
 from idpyoidc.client.oauth2.stand_alone_client import StandAloneClient
 from idpyoidc.server.user_authn.authn_context import UNSPECIFIED
 
-import satosa.logging_util as lu
 from satosa.backends.base import BackendModule
 from satosa.internal import AuthenticationInformation
 from satosa.internal import InternalData
+import satosa.logging_util as lu
 from ..exception import SATOSAAuthenticationError
 from ..exception import SATOSAError
 from ..response import Redirect
@@ -74,7 +74,7 @@ def register_endpoints(self):
         if not redirect_path:
             raise SATOSAError("Missing path in redirect uri")
         redirect_path = urlparse(redirect_path[0]).path
-        url_map.append(("^%s$" % redirect_path.lstrip("/"), self.response_endpoint))
+        url_map.append((f"^{redirect_path.lstrip('/')}$", self.response_endpoint))
         return url_map
 
     def response_endpoint(self, context, *args):

tests/satosa/backends/test_idpy_oidc.py

@@ -0,0 +1,207 @@
+import json
+import re
+import time
+from unittest.mock import Mock
+from urllib.parse import parse_qsl
+from urllib.parse import urlparse
+
+from cryptojwt.key_jar import build_keyjar
+from idpyoidc.client.defaults import DEFAULT_KEY_DEFS
+from idpyoidc.client.oauth2.stand_alone_client import StandAloneClient
+from idpyoidc.message.oidc import AuthorizationResponse
+from idpyoidc.message.oidc import IdToken
+from oic.oic import AuthorizationRequest
+import pytest
+import responses
+
+from satosa.backends.idpy_oidc import IdpyOIDCBackend
+from satosa.context import Context
+from satosa.internal import InternalData
+from satosa.response import Response
+
+ISSUER = "https://provider.example.com"
+CLIENT_ID = "test_client"
+CLIENT_BASE_URL = "https://client.test.com"
+NONCE = "the nonce"
+
+
+class TestIdpyOIDCBackend(object):
+    @pytest.fixture
+    def backend_config(self):
+        return {
+            "client": {
+                "base_url": CLIENT_BASE_URL,
+                "client_id": CLIENT_ID,
+                "client_type": "oidc",
+                "client_secret": "ZJYCqe3GGRvdrudKyZS0XhGv_Z45DuKhCUk0gBR1vZk",
+                "application_type": "web",
+                "application_name": "SATOSA Test",
+                "contacts": ["[email protected]"],
+                "response_types_supported": ["code"],
+                "response_type": "code id_token token",
+                "scope": "openid foo",
+                "key_conf": {"key_defs": DEFAULT_KEY_DEFS},
+                "jwks_uri": f"{CLIENT_BASE_URL}/jwks.json",
+                "provider_info": {
+                    "issuer": ISSUER,
+                    "authorization_endpoint": f"{ISSUER}/authn",
+                    "token_endpoint": f"{ISSUER}/token",
+                    "userinfo_endpoint": f"{ISSUER}/user",
+                    "jwks_uri": f"{ISSUER}/static/jwks"
+                }
+            }
+        }
+
+    @pytest.fixture
+    def internal_attributes(self):
+        return {
+            "attributes": {
+                "givenname": {"openid": ["given_name"]},
+                "mail": {"openid": ["email"]},
+                "edupersontargetedid": {"openid": ["sub"]},
+                "surname": {"openid": ["family_name"]}
+            }
+        }
+
+    @pytest.fixture(autouse=True)
+    @responses.activate
+    def create_backend(self, internal_attributes, backend_config):
+        base_url = backend_config['client']['base_url']
+        self.issuer_keys = build_keyjar(DEFAULT_KEY_DEFS)
+        with responses.RequestsMock() as rsps:
+            rsps.add(
+                responses.GET,
+                backend_config['client']['provider_info']['jwks_uri'],
+                body=self.issuer_keys.export_jwks_as_json(),
+                status=200,
+                content_type="application/json")
+
+            self.oidc_backend = IdpyOIDCBackend(Mock(), internal_attributes, backend_config,
+                                                base_url, "oidc")
+
+    @pytest.fixture
+    def userinfo(self):
+        return {
+            "given_name": "Test",
+            "family_name": "Devsson",
+            "email": "[email protected]",
+            "sub": "username"
+        }
+
+    def test_client(self, backend_config):
+        assert isinstance(self.oidc_backend.client, StandAloneClient)
+        # 3 signing keys. One RSA, one EC and one symmetric
+        assert len(self.oidc_backend.client.context.keyjar.get_signing_key()) == 3
+        assert self.oidc_backend.client.context.jwks_uri == backend_config['client']['jwks_uri']
+
+    def assert_expected_attributes(self, attr_map, user_claims, actual_attributes):
+        expected_attributes = {}
+        for out_attr, in_mapping in attr_map["attributes"].items():
+            expected_attributes[out_attr] = [user_claims[in_mapping["openid"][0]]]
+
+        assert actual_attributes == expected_attributes
+
+    def setup_token_endpoint(self, userinfo):
+        _client = self.oidc_backend.client
+        signing_key = self.issuer_keys.get_signing_key(key_type='RSA')[0]
+        signing_key.alg = "RS256"
+        id_token_claims = {
+            "iss": ISSUER,
+            "sub": userinfo["sub"],
+            "aud": CLIENT_ID,
+            "nonce": NONCE,
+            "exp": time.time() + 3600,
+            "iat": time.time()
+        }
+        id_token = IdToken(**id_token_claims).to_jwt([signing_key], algorithm=signing_key.alg)
+        token_response = {
+            "access_token": "SlAV32hkKG",
+            "token_type": "Bearer",
+            "refresh_token": "8xLOxBtZp8",
+            "expires_in": 3600,
+            "id_token": id_token
+        }
+        responses.add(responses.POST,
+                      _client.context.provider_info['token_endpoint'],
+                      body=json.dumps(token_response),
+                      status=200,
+                      content_type="application/json")
+
+    def setup_userinfo_endpoint(self, userinfo):
+        responses.add(responses.GET,
+                      self.oidc_backend.client.context.provider_info['userinfo_endpoint'],
+                      body=json.dumps(userinfo),
+                      status=200,
+                      content_type="application/json")
+
+    @pytest.fixture
+    def incoming_authn_response(self):
+        _context = self.oidc_backend.client.context
+        oidc_state = "my state"
+        _uri = _context.claims.get_usage("redirect_uris")[0]
+        _request = AuthorizationRequest(
+            redirect_uri=_uri,
+            response_type="code",
+            client_id=_context.get_client_id(),
+            scope=_context.claims.get_usage("scope"),
+            nonce=NONCE
+        )
+        _context.cstate.set(oidc_state, {"iss": _context.issuer})
+        _context.cstate.bind_key(NONCE, oidc_state)
+        _context.cstate.update(oidc_state, _request)
+
+        response = AuthorizationResponse(
+            code="F+R4uWbN46U+Bq9moQPC4lEvRd2De4o=",
+            state=oidc_state,
+            iss=_context.issuer,
+            nonce=NONCE
+        )
+        return response.to_dict()
+
+    def test_register_endpoints(self):
+        _uri = self.oidc_backend.client.context.claims.get_usage("redirect_uris")[0]
+        redirect_uri_path = urlparse(_uri).path.lstrip('/')
+        url_map = self.oidc_backend.register_endpoints()
+        regex, callback = url_map[0]
+        assert re.search(regex, redirect_uri_path)
+        assert callback == self.oidc_backend.response_endpoint
+
+    def test_translate_response_to_internal_response(self, userinfo):
+        internal_response = self.oidc_backend._translate_response(userinfo, ISSUER)
+        assert internal_response.subject_id == userinfo["sub"]
+        self.assert_expected_attributes(self.oidc_backend.internal_attributes, userinfo,
+                                        internal_response.attributes)
+
+    @responses.activate
+    def test_response_endpoint(self, context, userinfo, incoming_authn_response):
+        self.setup_token_endpoint(userinfo)
+        self.setup_userinfo_endpoint(userinfo)
+
+        response_context = Context()
+        response_context.request = incoming_authn_response
+        response_context.state = context.state
+
+        self.oidc_backend.response_endpoint(response_context)
+
+        args = self.oidc_backend.auth_callback_func.call_args[0]
+        assert isinstance(args[0], Context)
+        assert isinstance(args[1], InternalData)
+        self.assert_expected_attributes(self.oidc_backend.internal_attributes, userinfo,
+                                        args[1].attributes)
+
+    def test_start_auth_redirects_to_provider_authorization_endpoint(self, context):
+        _client = self.oidc_backend.client
+        auth_response = self.oidc_backend.start_auth(context, None)
+        assert isinstance(auth_response, Response)
+
+        login_url = auth_response.message
+        parsed = urlparse(login_url)
+        assert login_url.startswith(_client.context.provider_info["authorization_endpoint"])
+        auth_params = dict(parse_qsl(parsed.query))
+        assert auth_params["scope"] == " ".join(_client.context.claims.get_usage("scope"))
+        assert auth_params["response_type"] == _client.context.claims.get_usage("response_types")[0]
+        assert auth_params["client_id"] == _client.client_id
+        assert auth_params["redirect_uri"] == _client.context.claims.get_usage("redirect_uris")[0]
+        assert "state" in auth_params
+        assert "nonce" in auth_params
+