Partial commit.

Improved signing and added more testcases.

Author: HaToHo

src/saml2/entity.py

@@ -512,8 +512,14 @@ def _encrypt_assertion(self, encrypt_cert, sp_entity_id, response, node_xpath=No
         exception = None
         for _cert in _certs:
             try:
+                begin_cert = "-----BEGIN CERTIFICATE-----\n"
+                end_cert = "\n-----END CERTIFICATE-----\n"
+                if begin_cert not in _cert:
+                    _cert = "%s%s" % (begin_cert, _cert)
+                if end_cert not in _cert:
+                    _cert = "%s%s" % (_cert, end_cert)
                 _, cert_file = make_temp(_cert, decode=False)
-                response = cbxs.encrypt_assertion(response, self.sec.cert_file,
+                response = cbxs.encrypt_assertion(response, cert_file,
                                                   pre_encryption_part(), node_xpath=node_xpath)
                 return response
             except Exception as ex:
@@ -525,7 +531,7 @@ def _encrypt_assertion(self, encrypt_cert, sp_entity_id, response, node_xpath=No
     def _response(self, in_response_to, consumer_url=None, status=None,
                   issuer=None, sign=False, to_sign=None, sp_entity_id=None,
                   encrypt_assertion=False, encrypt_assertion_self_contained=False, encrypted_advice_attributes=False,
-                  encrypt_cert=None, encrypt_cert_assertion=None,sign_assertion=None, **kwargs):
+                  encrypt_cert_advice=None, encrypt_cert_assertion=None,sign_assertion=None, **kwargs):
         """ Create a Response.
             Encryption:
                 encrypt_assertion must be true for encryption to be performed. If encrypted_advice_attributes also is
@@ -598,7 +604,7 @@ def _response(self, in_response_to, consumer_url=None, status=None,
 
                 if to_sign_advice:
                     response = signed_instance_factory(response, self.sec, to_sign_advice)
-                response = self._encrypt_assertion(encrypt_cert, sp_entity_id, response, node_xpath=node_xpath)
+                response = self._encrypt_assertion(encrypt_cert_advice, sp_entity_id, response, node_xpath=node_xpath)
                 if encrypt_assertion:
                     response = response_from_string(response)
             if encrypt_assertion:

src/saml2/server.py

@@ -323,7 +323,7 @@ def _authn_response(self, in_response_to, consumer_url,
                         status=None, authn=None, issuer=None, policy=None,
                         sign_assertion=False, sign_response=False,
                         best_effort=False, encrypt_assertion=False,
-                        encrypt_cert=None, authn_statement=None,
+                        encrypt_cert_advice=None, encrypt_cert_assertion=None, authn_statement=None,
                         encrypt_assertion_self_contained=False, encrypted_advice_attributes=False):
         """ Create a response. A layer of indirection.
 
@@ -375,16 +375,17 @@ def _authn_response(self, in_response_to, consumer_url,
                                                          sign_response)
 
         to_sign = []
-        #if sign_assertion is not None and sign_assertion:
-        #    if assertion.advice and assertion.advice.assertion:
-        #        for tmp_assertion in assertion.advice.assertion:
-        #            tmp_assertion.signature = pre_signature_part(tmp_assertion.id, self.sec.my_cert, 1)
-        #            to_sign.append((class_name(tmp_assertion), tmp_assertion.id))
-        #    assertion.signature = pre_signature_part(assertion.id,
-        #                                             self.sec.my_cert, 1)
-            # Just the assertion or the response and the assertion ?
-        #    to_sign.append((class_name(assertion), assertion.id))
+        if not encrypt_assertion:
+            if sign_assertion:
+                assertion.signature = pre_signature_part(assertion.id, self.sec.my_cert, 1)
+                to_sign.append((class_name(assertion), assertion.id))
 
+        if not encrypted_advice_attributes:
+            if sign_assertion:
+                if assertion.advice and assertion.advice.assertion:
+                    for tmp_assertion in assertion.advice.assertion:
+                        tmp_assertion.signature = pre_signature_part(tmp_assertion.id, self.sec.my_cert, 1)
+                        to_sign.append((class_name(tmp_assertion), tmp_assertion.id))
 
         # Store which assertion that has been sent to which SP about which
         # subject.
@@ -400,7 +401,8 @@ def _authn_response(self, in_response_to, consumer_url,
 
         return self._response(in_response_to, consumer_url, status, issuer,
                               sign_response, to_sign,sp_entity_id=sp_entity_id, encrypt_assertion=encrypt_assertion,
-                              encrypt_cert=encrypt_cert,
+                              encrypt_cert_advice=encrypt_cert_advice,
+                              encrypt_cert_assertion=encrypt_cert_assertion,
                               encrypt_assertion_self_contained=encrypt_assertion_self_contained,
                               encrypted_advice_attributes=encrypted_advice_attributes,sign_assertion=sign_assertion,
                               **args)
@@ -477,8 +479,8 @@ def create_authn_response(self, identity, in_response_to, destination,
                               sp_entity_id, name_id_policy=None, userid=None,
                               name_id=None, authn=None, issuer=None,
                               sign_response=None, sign_assertion=None,
-                              encrypt_cert=None, encrypt_assertion=None,
-                              encrypt_assertion_self_contained=False,
+                              encrypt_cert_advice=None, encrypt_cert_assertion=None, encrypt_assertion=None,
+                              encrypt_assertion_self_contained=True,
                               encrypted_advice_attributes=False,
                               **kwargs):
         """ Constructs an AuthenticationResponse
@@ -523,17 +525,35 @@ def create_authn_response(self, identity, in_response_to, destination,
         if encrypt_assertion is None:
             encrypt_assertion = False
 
+
+        if encrypt_assertion_self_contained is None:
+            encrypt_assertion_self_contained = self.config.getattr("encrypt_assertion_self_contained", "idp")
+        if encrypt_assertion_self_contained is None:
+            encrypt_assertion_self_contained = True
+
+        if encrypted_advice_attributes is None:
+            encrypted_advice_attributes = self.config.getattr("encrypted_advice_attributes", "idp")
+        if encrypted_advice_attributes is None:
+            encrypted_advice_attributes = False
+
+        if encrypted_advice_attributes:
+            verify_encrypt_cert = self.config.getattr("verify_encrypt_cert_advice", "idp")
+            if verify_encrypt_cert is not None:
+                if encrypt_cert_advice is None:
+                    raise CertificateError("No SPCertEncType certificate for encryption contained in authentication "
+                   "request.")
+                if not verify_encrypt_cert(encrypt_cert_advice):
+                    raise CertificateError("Invalid certificate for encryption!")
+
+
         if encrypt_assertion:
-            if encrypt_cert is not None:
-                verify_encrypt_cert = self.config.getattr("verify_encrypt_cert", "idp")
-                if verify_encrypt_cert is not None:
-                    if not verify_encrypt_cert(encrypt_cert):
-                        raise CertificateError("Invalid certificate for encryption!")
-            else:
-                raise CertificateError("No SPCertEncType certificate for encryption contained in authentication "
-                                       "request.")
-        else:
-            encrypt_assertion = False
+            verify_encrypt_cert = self.config.getattr("verify_encrypt_cert_assertion", "idp")
+            if verify_encrypt_cert is not None:
+                if encrypt_cert_assertion is None:
+                    raise CertificateError("No SPCertEncType certificate for encryption contained in authentication "
+                                   "request.")
+                if not verify_encrypt_cert(encrypt_cert_assertion):
+                    raise CertificateError("Invalid certificate for encryption!")
 
         if not name_id:
             try:
@@ -593,7 +613,8 @@ def create_authn_response(self, identity, in_response_to, destination,
                                                 encrypt_assertion=encrypt_assertion,
                                                 encrypt_assertion_self_contained=encrypt_assertion_self_contained,
                                                 encrypted_advice_attributes=encrypted_advice_attributes,
-                                                encrypt_cert=encrypt_cert)
+                                                encrypt_cert_advice=encrypt_cert_advice,
+                                                encrypt_cert_assertion=encrypt_cert_assertion)
             return self._authn_response(in_response_to,  # in_response_to
                                         destination,  # consumer_url
                                         sp_entity_id,  # sp_entity_id
@@ -608,7 +629,8 @@ def create_authn_response(self, identity, in_response_to, destination,
                                         encrypt_assertion=encrypt_assertion,
                                         encrypt_assertion_self_contained=encrypt_assertion_self_contained,
                                         encrypted_advice_attributes=encrypted_advice_attributes,
-                                        encrypt_cert=encrypt_cert)
+                                        encrypt_cert_advice=encrypt_cert_advice,
+                                        encrypt_cert_assertion=encrypt_cert_assertion)
 
         except MissingValue as exc:
             return self.create_error_response(in_response_to, destination,