Remove asserts that are used in place of conditions (CZ-NIC#505)

Close CZ-NIC#504

src/oic/extension/message.py

@@ -8,6 +8,7 @@
 
 from oic.exception import InvalidRedirectUri
 from oic.exception import MissingPage
+from oic.exception import RegistrationError
 from oic.oauth2.message import OPTIONAL_LIST_OF_SP_SEP_STRINGS
 from oic.oauth2.message import OPTIONAL_LIST_OF_STRINGS
 from oic.oauth2.message import REQUIRED_LIST_OF_STRINGS
@@ -139,8 +140,8 @@ class RegistrationRequest(Message):
     }
 
     def verify(self, **kwargs):
-        if "initiate_login_uri" in self:
-            assert self["initiate_login_uri"].startswith("https:")
+        if "initiate_login_uri" in self and not self["initiate_login_uri"].startswith("https:"):
+            raise RegistrationError('initiate_login_uri is not https')
 
         if "redirect_uris" in self:
             for uri in self["redirect_uris"]:

src/oic/extension/provider.py

@@ -676,8 +676,10 @@ def code_grant_type(self, areq):
 
         # If redirect_uri was in the initial authorization request
         # verify that the one given here is the correct one.
-        if "redirect_uri" in _info:
-            assert areq["redirect_uri"] == _info["redirect_uri"]
+        if "redirect_uri" in _info and areq["redirect_uri"] != _info["redirect_uri"]:
+            logger.error('Redirect_uri mismatch')
+            err = TokenErrorResponse(error="unauthorized_client")
+            return Unauthorized(err.to_json(), content="application/json")
 
         issue_refresh = False
         if 'scope' in authzreq and 'offline_access' in authzreq['scope']:

src/oic/oauth2/provider.py

@@ -49,6 +49,7 @@
 from oic.utils.http_util import CookieDealer
 from oic.utils.http_util import Response
 from oic.utils.http_util import SeeOther
+from oic.utils.http_util import Unauthorized
 from oic.utils.http_util import make_cookie
 from oic.utils.sanitize import sanitize
 from oic.utils.sdb import AccessCodeUsed
@@ -244,18 +245,22 @@ def _verify_redirect_uri(self, areq):
                     # redirect_uri
                     if rquery:
                         for key, vals in rquery.items():
-                            assert key in _query
+                            if key not in _query:
+                                raise AssertionError()
                             for val in vals:
-                                assert val in _query[key]
+                                if val not in _query[key]:
+                                    raise AssertionError()
                     # and vice versa, every query component in the redirect_uri
                     # must be registered
                     if _query:
                         if rquery is None:
                             raise ValueError
                         for key, vals in _query.items():
-                            assert key in rquery
+                            if key not in rquery:
+                                raise AssertionError()
                             for val in vals:
-                                assert val in rquery[key]
+                                if val not in rquery[key]:
+                                    raise AssertionError()
                     match = True
                     break
             if not match:
@@ -801,8 +806,10 @@ def token_endpoint(self, authn="", **kwargs):
 
         # If redirect_uri was in the initial authorization request
         # verify that the one given here is the correct one.
-        if "redirect_uri" in _info:
-            assert areq["redirect_uri"] == _info["redirect_uri"]
+        if "redirect_uri" in _info and areq["redirect_uri"] != _info["redirect_uri"]:
+            logger.error('Redirect_uri mismatch')
+            err = TokenErrorResponse(error="unauthorized_client")
+            return Unauthorized(err.to_json(), content="application/json")
 
         try:
             _tinfo = _sdb.upgrade_to_token(areq["code"], issue_refresh=True)

src/oic/oic/message.py

@@ -440,8 +440,8 @@ def verify(self, **kwargs):
 
                 # verify that nothing is change in the original message
                 for key, val in oidr.items():
-                    if key in self:
-                        assert self[key] == val
+                    if key in self and self[key] != val:
+                        raise AssertionError()
 
                 # replace the JWT with the parsed and verified instance
                 self["request"] = oidr
@@ -584,8 +584,8 @@ class RegistrationRequest(Message):
     def verify(self, **kwargs):
         super(RegistrationRequest, self).verify(**kwargs)
 
-        if "initiate_login_uri" in self:
-            assert self["initiate_login_uri"].startswith("https:")
+        if "initiate_login_uri" in self and not self["initiate_login_uri"].startswith("https:"):
+            raise AssertionError()
 
         for param in ["request_object_encryption",
                       "id_token_encrypted_response",
@@ -597,11 +597,11 @@ def verify(self, **kwargs):
                     self[enc_param] = "A128CBC-HS256"
 
             # both or none
-            if enc_param in self:
-                assert alg_param in self
+            if enc_param in self and alg_param not in self:
+                raise AssertionError()
 
-        if "token_endpoint_auth_signing_alg" in self:
-            assert self["token_endpoint_auth_signing_alg"] != "none"
+        if "token_endpoint_auth_signing_alg" in self and self["token_endpoint_auth_signing_alg"] == "none":
+            raise AssertionError()
 
         return True
 
@@ -842,15 +842,17 @@ def verify(self, **kwargs):
         super(ProviderConfigurationResponse, self).verify(**kwargs)
 
         if "scopes_supported" in self:
-            assert "openid" in self["scopes_supported"]
+            if "openid" not in self["scopes_supported"]:
+                raise AssertionError()
             for scope in self["scopes_supported"]:
                 check_char_set(scope, SCOPE_CHARSET)
 
         parts = urlparse(self["issuer"])
         if parts.scheme != "https":
             raise SchemeError("Not HTTPS")
 
-        assert not parts.query and not parts.fragment
+        if parts.query or parts.fragment:
+            raise AssertionError()
 
         if any("code" in rt for rt in self[
                 "response_types_supported"]) and "token_endpoint" not in self:

src/oic/utils/authn/user.py

@@ -286,7 +286,8 @@ def __call__(self, cookie=None, end_point_index=0, **kwargs):
         return resp
 
     def _verify(self, pwd, user):
-        assert pwd == self.passwd[user], "Passwords don't match."
+        if self.passwd[user] != pwd:
+            raise AssertionError("Passwords don't match.")
 
     def verify(self, request, **kwargs):
         """
@@ -361,9 +362,7 @@ def __init__(self, srv, pwd, ttl=5):
         self.passwd = pwd
 
     def verify_password(self, user, password):
-        try:
-            assert password == self.passwd[user]
-        except (AssertionError, KeyError):
+        if not (user in self.passwd and password == self.passwd[user]):
             raise FailedAuthentication("Wrong password")
 
     def authenticated_as(self, cookie=None, authorization="", **kwargs):