Strict on which signing algorithm that can be used.

Author: rohe

src/oidcmsg/oidc/__init__.py

@@ -9,6 +9,7 @@
 import time
 
 from cryptojwt import as_unicode
+from cryptojwt.exception import UnsupportedAlgorithm
 from cryptojwt.jws.jws import factory as jws_factory
 from cryptojwt.jws.utils import left_hash
 from cryptojwt.jwt import JWT
@@ -241,7 +242,7 @@ def check_char_set(string, allowed):
 ID_TOKEN_VERIFY_ARGS = ['keyjar', 'verify', 'encalg', 'encenc', 'sigalg',
                         'issuer', 'allow_missing_kid', 'no_kid_issuer',
                         'trusting', 'skew', 'nonce_storage_time', 'client_id',
-                        'allow_sign_alg_none']
+                        'allow_sign_alg_none', 'allowed_sign_alg']
 
 CLAIMS_WITH_VERIFIED = ['id_token', 'id_token_hint', 'request']
 
@@ -282,11 +283,18 @@ def verify_id_token(msg, check_hash=False, claim='id_token', **kwargs):
             _allow_none = kwargs['allow_sign_alg_none']
         except KeyError:
             logger.info('Signing algorithm None not allowed')
-            return False
+            raise UnsupportedAlgorithm('Signing algorithm None not allowed')
         else:
             if not _allow_none:
                 logger.info('Signing algorithm None not allowed')
-                return False
+                raise UnsupportedAlgorithm('Signing algorithm None not allowed')
+    else:
+        if "allowed_sign_alg" in kwargs:
+            if _jws.jwt.headers['alg'] != kwargs["allowed_sign_alg"]:
+                _msg = "Wrong token signing algorithm, {} != {}".format(
+                    _jws.jwt.headers['alg'], kwargs["allowed_sign_alg"])
+                logger.error(_msg)
+                raise UnsupportedAlgorithm(_msg)
 
     _body = _jws.jwt.payload()
     if 'keyjar' in kwargs:

src/oidcmsg/oidc/session.py

@@ -1,21 +1,24 @@
 import logging
 
+from cryptojwt.exception import UnsupportedAlgorithm
+
 from oidcmsg.time_util import utc_time_sans_frac
-from ..exception import MessageException, NotForMe
+from ..exception import MessageException
+from ..exception import NotForMe
 from ..message import Message
 from ..message import REQUIRED_LIST_OF_STRINGS
 from ..message import SINGLE_OPTIONAL_STRING
 from ..message import SINGLE_REQUIRED_INT
 from ..message import SINGLE_REQUIRED_JSON
 from ..message import SINGLE_REQUIRED_STRING
 from ..oauth2 import ResponseMessage
-from ..oidc import clear_verified_claims, verify_id_token
-from ..oidc import verified_claim_name
-from ..oidc import IdToken
 from ..oidc import ID_TOKEN_VERIFY_ARGS
+from ..oidc import IdToken
 from ..oidc import MessageWithIdToken
 from ..oidc import SINGLE_OPTIONAL_IDTOKEN
-
+from ..oidc import clear_verified_claims
+from ..oidc import verified_claim_name
+from ..oidc import verify_id_token
 
 logger = logging.getLogger(__name__)
 
@@ -25,7 +28,7 @@ class RefreshSessionRequest(MessageWithIdToken):
     c_param.update({
         "redirect_url": SINGLE_REQUIRED_STRING,
         "state": SINGLE_REQUIRED_STRING
-        })
+    })
 
 
 class RefreshSessionResponse(MessageWithIdToken, ResponseMessage):
@@ -47,7 +50,7 @@ class EndSessionRequest(Message):
         "id_token_hint": SINGLE_OPTIONAL_IDTOKEN,
         "post_logout_redirect_uri": SINGLE_OPTIONAL_STRING,
         "state": SINGLE_OPTIONAL_STRING
-        }
+    }
 
     def verify(self, **kwargs):
         super(EndSessionRequest, self).verify(**kwargs)
@@ -111,7 +114,7 @@ def verify(self, **kwargs):
             raise ValueError('Wrong member value in "events"')
 
         # There must be either a 'sub' or a 'sid', and may contain both
-        if not('sub' in self or 'sid' in self):
+        if not ('sub' in self or 'sid' in self):
             raise ValueError('There MUST be either a "sub" or a "sid"')
 
         try:
@@ -141,6 +144,12 @@ def verify(self, **kwargs):
             if self['iat'] > (_now + _skew):
                 raise ValueError('Invalid issued_at time')
 
+        _allowed = kwargs.get("allowed_sign_alg")
+        if _allowed and self.jws_header['alg'] != _allowed:
+            _msg = "Wrong token signing algorithm, {} != {}".format(
+                self.jws_header['alg'], kwargs["allowed_sign_alg"])
+            raise UnsupportedAlgorithm(_msg)
+
         return True
 
 
@@ -155,7 +164,7 @@ class BackChannelLogoutRequest(Message):
 
     c_param = {
         "logout_token": SINGLE_REQUIRED_STRING
-        }
+    }
 
     def verify(self, **kwargs):
         super(BackChannelLogoutRequest, self).verify(**kwargs)