Defining capabilities not anymore done in one central place.

Most of it is done per endpoint.
Defined default capabilities per endpoint.
Some capabilities are computed at start time since they are configuration dependent.
Refactored.
Bumped version.

Author: rohe

src/oidcendpoint/__init__.py

@@ -6,7 +6,7 @@
 except ImportError:
     import random as rnd
 
-__version__ = "0.10.1"
+__version__ = "0.11.0"
 
 
 DEF_SIGN_ALG = {

src/oidcendpoint/cookie.py

@@ -24,7 +24,7 @@
 
 __author__ = "Roland Hedberg"
 
-logger = logging.getLogger(__name__)
+LOGGER = logging.getLogger(__name__)
 
 CORS_HEADERS = [
     ("Access-Control-Allow-Origin", "*"),
@@ -466,12 +466,15 @@ def append_cookie(
 
 def compute_session_state(opbs, salt, client_id, redirect_uri):
     """
+    Computes a session state value.
+    This value is later used during session management to check whether
+    the log in state has changed.
 
-    :param opbs:
+    :param opbs: Cookie value
     :param salt:
     :param client_id:
     :param redirect_uri:
-    :return:
+    :return: Session state value
     """
     parsed_uri = urlparse(redirect_uri)
     rp_origin_url = "{uri.scheme}://{uri.netloc}".format(uri=parsed_uri)

src/oidcendpoint/endpoint.py

@@ -1,6 +1,9 @@
 import logging
+from functools import cmp_to_key
 from urllib.parse import urlparse
 
+from cryptojwt import jwe
+from cryptojwt.jws.jws import SIGNER_ALGS
 from oidcmsg.exception import MissingRequiredAttribute
 from oidcmsg.exception import MissingRequiredValue
 from oidcmsg.message import Message
@@ -51,6 +54,72 @@ def fragment_encoding(return_type):
         return True
 
 
+ALG_SORT_ORDER = {"RS": 0, "ES": 1, "HS": 2, "PS": 3, "no": 4}
+
+
+def sort_sign_alg(alg1, alg2):
+    if ALG_SORT_ORDER[alg1[0:2]] < ALG_SORT_ORDER[alg2[0:2]]:
+        return -1
+
+    if ALG_SORT_ORDER[alg1[0:2]] > ALG_SORT_ORDER[alg2[0:2]]:
+        return 1
+
+    if alg1 < alg2:
+        return -1
+
+    if alg1 > alg2:
+        return 1
+
+    return 0
+
+
+def assign_algorithms(typ):
+    if typ == "signing_alg":
+        # Pick supported signing algorithms from crypto library
+        # Sort order RS, ES, HS, PS
+        sign_algs = list(SIGNER_ALGS.keys())
+        return sorted(sign_algs, key=cmp_to_key(sort_sign_alg))
+    elif typ == "encryption_alg":
+        return jwe.SUPPORTED["alg"]
+    elif typ == "encryption_enc":
+        return jwe.SUPPORTED["enc"]
+
+
+def construct_provider_info(default_capabilities, **kwargs):
+    if default_capabilities is not None:
+        provider_info = {}
+        for attr, default_val in default_capabilities.items():
+            try:
+                _proposal = kwargs[attr]
+            except KeyError:
+                if default_val is not None:
+                    provider_info[attr] = default_val
+                elif "signing_alg_values_supported" in attr:
+                    provider_info[attr] = assign_algorithms("signing_alg")
+                elif "encryption_alg_values_supported" in attr:
+                    provider_info[attr] = assign_algorithms("encryption_alg")
+                elif "encryption_enc_values_supported" in attr:
+                    provider_info[attr] = assign_algorithms("encryption_enc")
+            else:
+                _permitted = None
+
+                if "signing_alg_values_supported" in attr:
+                    _permitted = set(assign_algorithms("signing_alg"))
+                elif "encryption_alg_values_supported" in attr:
+                    _permitted = set(assign_algorithms("encryption_alg"))
+                elif "encryption_enc_values_supported" in attr:
+                    _permitted = set(assign_algorithms("encryption_enc"))
+
+                if _permitted and not _permitted.issuperset(set(_proposal)):
+                    raise ValueError(
+                        "Proposed set of values outside set of permitted ({})".__format__(attr))
+
+                provider_info[attr] = _proposal
+        return provider_info
+    else:
+        return None
+
+
 class Endpoint(object):
     request_cls = Message
     response_cls = Message
@@ -63,6 +132,7 @@ class Endpoint(object):
     response_format = "json"
     response_placement = "body"
     client_authn_method = ""
+    default_capabilities = None
 
     def __init__(self, endpoint_context, **kwargs):
         self.endpoint_context = endpoint_context
@@ -71,7 +141,14 @@ def __init__(self, endpoint_context, **kwargs):
         self.post_parse_request = []
         self.kwargs = kwargs
         self.full_path = ""
-        self.provider_info = None
+
+        if "client_authn_method" in kwargs:
+            self.client_authn_method = kwargs["client_authn_method"]
+        elif self.default_capabilities is not None:
+            if "client_authn_method" in self.default_capabilities:
+                self.client_authn_method = self.default_capabilities["client_authn_method"]
+
+        self.provider_info = construct_provider_info(self.default_capabilities, **kwargs)
 
     def parse_request(self, request, auth=None, **kwargs):
         """

src/oidcendpoint/endpoint_context.py

@@ -1,11 +1,7 @@
-import copy
 import logging
 import os
-from functools import cmp_to_key
 
 import requests
-from cryptojwt import jwe
-from cryptojwt.jws.jws import SIGNER_ALGS
 from cryptojwt.key_jar import KeyJar
 from cryptojwt.key_jar import init_key_jar
 from jinja2 import Environment
@@ -16,7 +12,6 @@
 from oidcendpoint import rndstr
 from oidcendpoint import util
 from oidcendpoint.client_authn import CLIENT_AUTHN_METHOD
-from oidcendpoint.exception import ConfigurationError
 from oidcendpoint.id_token import IDToken
 from oidcendpoint.session import create_session_db
 from oidcendpoint.sso_db import SSODb
@@ -28,55 +23,6 @@
 
 LOGGER = logging.getLogger(__name__)
 
-CAPABILITIES = {
-    "response_types_supported": [
-        "code",
-        "token",
-        "id_token",
-        "code token",
-        "code id_token",
-        "id_token token",
-        "code id_token token",
-        "none",
-    ],
-    "token_endpoint_auth_methods_supported": [
-        "client_secret_post",
-        "client_secret_basic",
-        "client_secret_jwt",
-        "private_key_jwt",
-    ],
-    "response_modes_supported": ["query", "fragment", "form_post"],
-    "subject_types_supported": ["public", "pairwise"],
-    "grant_types_supported": [
-        "authorization_code",
-        "implicit",
-        "urn:ietf:params:oauth:grant-type:jwt-bearer",
-        "refresh_token",
-    ],
-    "claim_types_supported": ["normal", "aggregated", "distributed"],
-    "claims_parameter_supported": True,
-    "request_parameter_supported": True,
-    "request_uri_parameter_supported": True,
-}
-
-SORT_ORDER = {"RS": 0, "ES": 1, "HS": 2, "PS": 3, "no": 4}
-
-
-def sort_sign_alg(alg1, alg2):
-    if SORT_ORDER[alg1[0:2]] < SORT_ORDER[alg2[0:2]]:
-        return -1
-
-    if SORT_ORDER[alg1[0:2]] > SORT_ORDER[alg2[0:2]]:
-        return 1
-
-    if alg1 < alg2:
-        return -1
-
-    if alg1 > alg2:
-        return 1
-
-    return 0
-
 
 def add_path(url, path):
     if url.endswith("/"):
@@ -375,9 +321,9 @@ def do_endpoints(self, conf):
             if endpoint in ["webfinger", "provider_info"]:
                 continue
 
-            _cap[endpoint_instance.endpoint_name] = "{}".format(
-                endpoint_instance.endpoint_path
-            )
+            if endpoint_instance.endpoint_name:
+                _cap[endpoint_instance.endpoint_name] = endpoint_instance.full_path
+
         return _cap
 
     def do_authz(self, conf):
@@ -388,90 +334,11 @@ def do_authz(self, conf):
         else:
             self.authz = init_service(authz_spec, self)
 
-    def package_capabilities(self):
-        _provider_info = copy.deepcopy(CAPABILITIES)
-        _provider_info["issuer"] = self.issuer
-        _provider_info["version"] = "3.0"
-
-        _claims = []
-        for _cl in self.scope2claims.values():
-            _claims.extend(_cl)
-        _provider_info["claims_supported"] = list(set(_claims))
-
-        _scopes = list(self.scope2claims.keys())
-        _provider_info["scopes_supported"] = _scopes
-
-        # Sort order RS, ES, HS, PS
-        sign_algs = list(SIGNER_ALGS.keys())
-        sign_algs = sorted(sign_algs, key=cmp_to_key(sort_sign_alg))
-
-        for typ in ["userinfo", "id_token", "request_object"]:
-            _provider_info["%s_signing_alg_values_supported" % typ] = sign_algs
-
-        # Remove 'none' for token_endpoint_auth_signing_alg_values_supported
-        # since it is not allowed
-        sign_algs = sign_algs[:]
-        sign_algs.remove("none")
-        _provider_info["token_endpoint_auth_signing_alg_values_supported"] = sign_algs
-
-        algs = jwe.SUPPORTED["alg"]
-        for typ in ["userinfo", "id_token", "request_object"]:
-            _provider_info["%s_encryption_alg_values_supported" % typ] = algs
-
-        encs = jwe.SUPPORTED["enc"]
-        for typ in ["userinfo", "id_token", "request_object"]:
-            _provider_info["%s_encryption_enc_values_supported" % typ] = encs
-
-        # acr_values
-        if self.authn_broker:
-            acr_values = self.authn_broker.get_acr_value_string()
-            if acr_values is not None:
-                _provider_info["acr_values_supported"] = acr_values
-
-        return _provider_info
-
-    @staticmethod
-    def get_wanted(key, val):
-        if isinstance(val, str):
-            _wanted = {val}
-        else:
-            try:
-                _wanted = set(val)
-            except TypeError:
-                if key == "response_types_supported":
-                    _wanted = set()
-                    for _v in val:
-                        _v.sort()
-                        _wanted.add(" ".join(_v))
-                else:
-                    raise
-            else:
-                _wanted = set()
-                for _v in val:
-                    _vals = _v.split(" ")
-                    _vals.sort()
-                    _wanted.add(" ".join(_vals))
-        return _wanted
-
-    def verify_allowed(self, allowed, key, val, pinfo, not_supported):
-        if isinstance(allowed, bool):
-            if allowed is False:
-                if val is True:
-                    not_supported[key] = True
-            else:
-                pinfo[key] = val
-        elif isinstance(allowed, str):
-            if val != allowed:
-                not_supported[key] = val
-        elif isinstance(allowed, list):
-            _wanted = self.get_wanted(key, val)
-
-            _allowed = set(allowed)
-
-            if (_wanted & _allowed) == _wanted:
-                pinfo[key] = list(_wanted)
-            else:
-                not_supported[key] = list(_wanted - _allowed)
+    def claims_supported(self):
+        _claims = set()
+        for scope, claims in self.scope2claims.items():
+            _claims.update(set(claims))
+        return list(_claims)
 
     def create_providerinfo(self, capabilities):
         """
@@ -481,28 +348,20 @@ def create_providerinfo(self, capabilities):
         :return:
         """
 
-        _pinfo = self.package_capabilities()
-        not_supported = {}
-        for key, val in capabilities.items():
-            try:
-                allowed = _pinfo[key]
-            except KeyError:
-                _pinfo[key] = val
-            else:
-                self.verify_allowed(allowed, key, val, _pinfo, not_supported)
+        _provider_info = capabilities
+        _provider_info["issuer"] = self.issuer
+        _provider_info["version"] = "3.0"
 
-        if not_supported:
-            _msg = "Server doesn't support the following features: {}".format(
-                not_supported
-            )
-            LOGGER.error(_msg)
-            raise ConfigurationError(_msg)
+        # acr_values
+        if self.authn_broker:
+            acr_values = self.authn_broker.get_acr_values()
+            if acr_values is not None:
+                _provider_info["acr_values_supported"] = acr_values
 
         if self.jwks_uri and self.keyjar:
-            _pinfo["jwks_uri"] = self.jwks_uri
+            _provider_info["jwks_uri"] = self.jwks_uri
 
-        for name, instance in self.endpoint.items():
-            if name not in ["webfinger", "provider_info"]:
-                _pinfo[instance.endpoint_name] = instance.full_path
+        _provider_info.update(self.idtoken.provider_info)
+        _provider_info['claims_supported'] = self.claims_supported()
 
-        return _pinfo
+        return _provider_info