Replace assert with proper checks

Signed-off-by: Ivan Kanakarakis <ivan.kanak@gmail.com>

Author: c00kiemon5ter

src/saml2/assertion.py

@@ -296,9 +296,7 @@ def post_entity_categories(maps, **kwargs):
                         else:
                             attrs = atlist
                         for _key in key:
-                            try:
-                                assert _key in ecs
-                            except AssertionError:
+                            if _key not in ecs:
                                 attrs = []
                                 break
                     elif key in ecs:

src/saml2/authn_context/__init__.py

@@ -90,7 +90,9 @@ def add(self, spec, method, level=0, authn_authority="", reference=None):
         if _ref is None:
             _ref = str(self.next)
 
-        assert _ref not in self.db["info"]
+        if _ref in self.db["info"]:
+            raise Exception("Internal error: reference is not unique")
+
         self.db["info"][_ref] = _info
         try:
             self.db["key"][key].append(_ref)

src/saml2/client.py

@@ -75,7 +75,12 @@ def prepare_for_authenticate(
                 response_binding=response_binding,
                 **kwargs)
 
-        assert negotiated_binding == binding
+        if negotiated_binding != binding:
+            raise ValueError(
+                "Negotiated binding '{}' does not match binding to use '{}'".format(
+                    negotiated_binding, binding
+                )
+            )
 
         return reqid, info
 

src/saml2/client_base.py

@@ -646,8 +646,10 @@ def create_name_id_mapping_request(self, name_id_policy,
         :return:
         """
 
-        # One of them must be present
-        assert name_id or base_id or encrypted_id
+        if not name_id and not base_id and not encrypted_id:
+            raise ValueError(
+                "At least one of name_id, base_id or encrypted_id must be present."
+            )
 
         if name_id:
             return self._message(NameIDMappingRequest, destination, message_id,

src/saml2/discovery.py

@@ -24,10 +24,10 @@ def parse_discovery_service_request(self, url="", query=""):
 
         # verify
 
-        for key in ["isPassive", "return", "returnIDParam", "policy",
-                    'entityID']:
+        for key in ["isPassive", "return", "returnIDParam", "policy", 'entityID']:
             try:
-                assert len(dsr[key]) == 1
+                if len(dsr[key]) != 1:
+                    raise Exception("Invalid DS request keys: {k}".format(k=key))
                 dsr[key] = dsr[key][0]
             except KeyError:
                 pass
@@ -37,20 +37,27 @@ def parse_discovery_service_request(self, url="", query=""):
             if part.query:
                 qp = parse.parse_qs(part.query)
                 if "returnIDParam" in dsr:
-                    assert dsr["returnIDParam"] not in qp.keys()
+                    if dsr["returnIDParam"] in qp.keys():
+                        raise Exception(
+                            "returnIDParam value should not be in the query params"
+                        )
                 else:
-                    assert "entityID" not in qp.keys()
+                    if "entityID" in qp.keys():
+                        raise Exception("entityID should not be in the query params")
         else:
             # If metadata not used this is mandatory
             raise VerificationError("Missing mandatory parameter 'return'")
 
         if "policy" not in dsr:
             dsr["policy"] = IDPDISC_POLICY
 
-        try:
-            assert dsr["isPassive"] in ["true", "false"]
-        except KeyError:
-            pass
+        is_passive = dsr.get("isPassive")
+        if is_passive not in ["true", "false"]:
+            raise ValueError(
+                "Invalid value '{v}' for attribute '{attr}'".format(
+                    v=is_passive, attr="isPassive"
+                )
+            )
 
         if "isPassive" in dsr and dsr["isPassive"] == "true":
             dsr["isPassive"] = True
@@ -93,10 +100,6 @@ def verify_sp_in_metadata(self, entity_id):
 
     def verify_return(self, entity_id, return_url):
         for endp in self.metadata.discovery_response(entity_id):
-            try:
-                assert return_url.startswith(endp["location"])
-            except AssertionError:
-                pass
-            else:
+            if not return_url.startswith(endp["location"]):
                 return True
         return False

src/saml2/ecp_client.py

@@ -136,7 +136,14 @@ def phase2(self, authn_request, rc_url, idp_entity_id, headers=None,
         logger.debug("[P2] IdP response dict: %s", respdict)
 
         idp_response = respdict["body"]
-        assert idp_response.c_tag == "Response"
+
+        expected_tag = "Response"
+        if idp_response.c_tag != expected_tag:
+            raise ValueError(
+                "Invalid Response tag '{invalid}' should be '{valid}'".format(
+                    invalid=idp_response.c_tag, valid=expected_tag
+                )
+            )
 
         logger.debug("[P2] IdP AUTHN response: %s", idp_response)
 
@@ -165,7 +172,14 @@ def parse_sp_ecp_response(respdict):
 
         # AuthnRequest in the body or not
         authn_request = respdict["body"]
-        assert authn_request.c_tag == "AuthnRequest"
+
+        expected_tag = "AuthnRequest"
+        if authn_request.c_tag != expected_tag:
+            raise ValueError(
+                "Invalid AuthnRequest tag '{invalid}' should be '{valid}'".format(
+                    invalid=authn_request.c_tag, valid=expected_tag
+                )
+            )
 
         # ecp.RelayState among headers
         _relay_state = None

src/saml2/entity.py

@@ -848,7 +848,7 @@ def _parse_request(self, enc_request, request_cls, service, binding):
         _log_debug("Loaded request")
 
         if _request:
-            _request = _request.verify()
+            _request.verify()
             _log_debug("Verified request")
 
         if not _request:
@@ -1192,14 +1192,14 @@ def _parse_response(self, xmlstr, response_cls, service, binding,
             response.require_signature = True
             # Verify that the assertion is syntactically correct and the
             # signature on the assertion is correct if present.
-            response = response.verify(keys)
+            response.verify(keys)
         except SignatureError as err:
             if require_signature:
                 logger.error("Signature Error: %s", err)
                 raise
             else:
                 response.require_signature = require_signature
-                response = response.verify(keys)
+                response.verify(keys)
         else:
             assertions_are_signed = True
         finally:
@@ -1260,7 +1260,13 @@ def artifact2destination(self, artifact, descriptor):
 
         _art = base64.b64decode(artifact)
 
-        assert _art[:2] == ARTIFACT_TYPECODE
+        typecode = _art[:2]
+        if typecode != ARTIFACT_TYPECODE:
+            raise ValueError(
+                "Invalid artifact typecode '{invalid}' should be {valid}".format(
+                    invalid=typecode, valid=ARTIFACT_TYPECODE
+                )
+            )
 
         try:
             endpoint_index = str(int(_art[2:4]))

src/saml2/mdstore.py

@@ -405,9 +405,7 @@ def entity_categories(self, entity_id):
         return res
 
     def __eq__(self, other):
-        try:
-            assert isinstance(other, MetaData)
-        except AssertionError:
+        if not isinstance(other, MetaData):
             return False
 
         if len(self.entity) != len(other.entity):
@@ -417,9 +415,7 @@ def __eq__(self, other):
             return False
 
         for key, item in self.entity.items():
-            try:
-                assert item == other[key]
-            except AssertionError:
+            if item != other[key]:
                 return False
 
         return True

src/saml2/pack.py

@@ -179,7 +179,10 @@ def http_redirect_message(message, location, relay_state="", typ="SAMLRequest",
 
     if signer:
         # sigalgs, should be one defined in xmldsig
-        assert sigalg in [b for a, b in SIG_ALLOWED_ALG]
+        if sigalg not in [long_name for short_name, long_name in SIG_ALLOWED_ALG]:
+            raise Exception(
+                "Signature algo not in allowed list: {algo}".format(algo=sigalg)
+            )
         args["SigAlg"] = sigalg
 
         string = "&".join([urlencode({k: args[k]})
@@ -269,7 +272,14 @@ def parse_soap_enveloped_saml(text, body_class, header_class=None):
     :return: header parts and body as saml.samlbase instances
     """
     envelope = defusedxml.ElementTree.fromstring(text)
-    assert envelope.tag == '{%s}Envelope' % NAMESPACE
+
+    envelope_tag = "{%s}Envelope" % NAMESPACE
+    if envelope.tag != envelope_tag:
+        raise ValueError(
+            "Invalid envelope tag '{invalid}' should be '{valid}'".format(
+                invalid=envelope.tag, valid=envelope_tag
+            )
+        )
 
     # print(len(envelope))
     body = None

src/saml2/request.py

@@ -80,15 +80,21 @@ def issue_instant_ok(self):
         return issued_at > lower and issued_at < upper
 
     def _verify(self):
-        assert self.message.version == "2.0"
+        valid_version = "2.0"
+        if self.message.version != valid_version:
+            raise VersionMismatch(
+                "Invalid version {invalid} should be {valid}".format(
+                    invalid=self.message.version, valid=valid_version
+                )
+            )
+
         if self.message.destination and self.receiver_addrs and \
                 self.message.destination not in self.receiver_addrs:
-            logger.error("%s not in %s", self.message.destination,
-                                         self.receiver_addrs)
+            logger.error("%s not in %s", self.message.destination, self.receiver_addrs)
             raise OtherError("Not destined for me!")
 
-        assert self.issue_instant_ok()
-        return self
+        valid = self.issue_instant_ok()
+        return valid
 
     def loads(self, xmldata, binding, origdoc=None, must=None,
               only_valid_cert=False):