Resolve sign_alg and digest_alg wherever pre_signature_part is called

Signed-off-by: Ivan Kanakarakis <[email protected]>

Author: c00kiemon5ter

src/saml2/entity.py

@@ -772,6 +772,10 @@ def _response(
                 and len(response.assertion.advice.assertion) == 1
             )
         ):
+            # XXX sig/digest-allowed should be configurable
+            sign_alg = sign_alg or self.signing_algorithm
+            digest_alg = digest_alg or self.digest_algorithm
+
             # XXX part-A (common) prepare sign response
             if sign:
                 response.signature = pre_signature_part(

src/saml2/server.py

@@ -414,8 +414,8 @@ def setup_assertion(
                 **kwargs)
         return assertion
 
-    # XXX calls pre_signature_part
-    # XXX > _response
+    # XXX DONE calls pre_signature_part
+    # XXX calls _response
     def _authn_response(
         self,
         in_response_to,
@@ -479,7 +479,6 @@ def _authn_response(
         if farg is None:
             assertion_args = {}
 
-        args = {}
         # if identity:
         _issuer = self._issuer(issuer)
 
@@ -517,13 +516,21 @@ def _authn_response(
         to_sign = []
         if not encrypt_assertion:
             if sign_assertion:
+                # XXX self.signing_algorithm self.digest_algorithm defined by entity
+                # XXX this should be handled through entity.py
+                # XXX sig/digest-allowed should be configurable
+                sign_alg = sign_alg or self.signing_algorithm
+                digest_alg = digest_alg or self.digest_algorithm
+
                 assertion.signature = pre_signature_part(
-                    assertion.id, self.sec.my_cert, 2, sign_alg=sign_alg, digest_alg=digest_alg
+                    assertion.id,
+                    self.sec.my_cert,
+                    2,
+                    sign_alg=sign_alg,
+                    digest_alg=digest_alg,
                 )
                 to_sign.append((class_name(assertion), assertion.id))
 
-        args["assertion"] = assertion
-
         if (self.support_AssertionIDRequest() or self.support_AuthnQuery()):
             self.session_db.store_assertion(assertion, to_sign)
 
@@ -544,7 +551,7 @@ def _authn_response(
             pefim=pefim,
             sign_alg=sign_alg,
             digest_alg=digest_alg,
-            **args,
+            assertion=assertion,
         )
 
     # ------------------------------------------------------------------------
@@ -868,7 +875,7 @@ def create_authn_request_response(
             digest_alg=digest_alg,
         )
 
-    # XXX calls pre_signature_part without ensuring sign_alg/digest_alg
+    # XXX DONE calls pre_signature_part
     # XXX DONE idp create > [...]
     def create_assertion_id_request_response(
         self, assertion_id, sign=None, sign_alg=None, digest_alg=None, **kwargs
@@ -880,7 +887,12 @@ def create_assertion_id_request_response(
 
         if to_sign:
             if assertion.signature is None:
-                # XXX calls pre_signature_part without ensuring sign_alg/digest_alg
+                # XXX self.signing_algorithm self.digest_algorithm defined by entity
+                # XXX this should be handled through entity.py
+                # XXX sig/digest-allowed should be configurable
+                sign_alg = sign_alg or self.signing_algorithm
+                digest_alg = digest_alg or self.digest_algorithm
+
                 assertion.signature = pre_signature_part(
                     assertion.id,
                     self.sec.my_cert,

tests/test_52_default_sign_alg.py

@@ -42,17 +42,8 @@ def get_ava(assertion):
 
 
 class TestSignedResponse():
-
     def setup_class(self):
         self.server = Server("idp_conf")
-        sign_alg = Mock()
-        sign_alg.return_value = ds.SIG_RSA_SHA512
-        digest_alg = Mock()
-        digest_alg.return_value = ds.DIGEST_SHA512
-        self.restet_default = ds.DefaultSignature
-        ds.DefaultSignature = MagicMock()
-        ds.DefaultSignature().get_sign_alg = sign_alg
-        ds.DefaultSignature().get_digest_alg = digest_alg
         conf = config.SPConfig()
         conf.load_file("server_conf")
         self.client = client.Saml2Client(conf)
@@ -62,7 +53,6 @@ def setup_class(self):
                "mail": ["[email protected]"], "title": "The man"}
 
     def teardown_class(self):
-        ds.DefaultSignature = self.restet_default
         self.server.close()
 
     def verify_assertion(self, assertion):
@@ -76,7 +66,6 @@ def verify_assertion(self, assertion):
                 'surName': ['Jeter'], 'title': ['The man']}
 
     def test_signed_response(self):
-
         print(ds.DefaultSignature().get_digest_alg())
         name_id = self.server.ident.transient_nameid(
             "urn:mace:example.com:saml:roland:sp", "id12")
@@ -96,11 +85,10 @@ def test_signed_response(self):
         assert signed_resp
 
         sresponse = response_from_string(signed_resp)
-        assert ds.SIG_RSA_SHA512 in str(sresponse), "Not correctly signed!"
-        assert ds.DIGEST_SHA512 in str(sresponse), "Not correctly signed!"
+        assert ds.SIG_RSA_SHA1 in str(sresponse), "Not correctly signed!"
+        assert ds.DIGEST_SHA1 in str(sresponse), "Not correctly signed!"
 
     def test_signed_response_1(self):
-
         signed_resp = self.server.create_authn_response(
             self.ava,
             "id12",  # in_response_to
@@ -112,15 +100,15 @@ def test_signed_response_1(self):
         )
 
         sresponse = response_from_string(signed_resp)
-        assert ds.SIG_RSA_SHA512 in str(sresponse), "Not correctly signed!"
-        assert ds.DIGEST_SHA512 in str(sresponse), "Not correctly signed!"
+        assert ds.SIG_RSA_SHA1 in str(sresponse), "Not correctly signed!"
+        assert ds.DIGEST_SHA1 in str(sresponse), "Not correctly signed!"
         valid = self.server.sec.verify_signature(signed_resp,
                                                  self.server.config.cert_file,
                                                  node_name='urn:oasis:names:tc:SAML:2.0:protocol:Response',
                                                  node_id=sresponse.id)
         assert valid
-        assert ds.SIG_RSA_SHA512 in str(sresponse.assertion[0]), "Not correctly signed!"
-        assert ds.DIGEST_SHA512 in str(sresponse.assertion[0]), "Not correctly signed!"
+        assert ds.SIG_RSA_SHA1 in str(sresponse.assertion[0]), "Not correctly signed!"
+        assert ds.DIGEST_SHA1 in str(sresponse.assertion[0]), "Not correctly signed!"
         valid = self.server.sec.verify_signature(signed_resp,
                                                  self.server.config.cert_file,
                                                  node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion',
@@ -130,7 +118,6 @@ def test_signed_response_1(self):
         self.verify_assertion(sresponse.assertion)
 
     def test_signed_response_2(self):
-
         signed_resp = self.server.create_authn_response(
             self.ava,
             "id12",  # in_response_to
@@ -161,6 +148,7 @@ def test_signed_response_2(self):
 
         self.verify_assertion(sresponse.assertion)
 
+
 if __name__ == "__main__":
     ts = TestSignedResponse()
     ts.setup_class()