def test_exception_sign_verify_with_cert_from_instance(self):
        assertion = factory(saml.Assertion,
                            version="2.0",
                            id="11100",
                            issue_instant="2009-10-30T13:20:28Z",
                            #signature= sigver.pre_signature_part("11100",
                            # self.sec.my_cert),
                            attribute_statement=do_attribute_statement({
                                ("", "", "surName"): ("Foo", ""),
                                ("", "", "givenName"): ("Bar", ""),
                            })
        )

        response = factory(samlp.Response,
                           assertion=assertion,
                           id="22222",
                           signature=sigver.pre_signature_part("22222",
                                                               self.sec
                                                               .my_cert))

        to_sign = [(class_name(response), response.id)]

        s_response = sigver.signed_instance_factory(response, self.sec, to_sign)

        response2 = response_from_string(s_response)
        # Change something that should make everything fail
        response2.id = "23456"
        raises(sigver.SignatureError, self.sec._check_signature,
               s_response, response2, class_name(response2))

    def test_sign_response_2(self):
        assertion2 = factory( saml.Assertion,
            version= "2.0",
            id= "11122",
            issue_instant= "2009-10-30T13:20:28Z",
            signature= sigver.pre_signature_part("11122", self.sec.my_cert),
            attribute_statement=do_attribute_statement({
                    ("","","surName"): ("Fox",""),
                    ("","","givenName") :("Bear",""),
                })
            )
        response = factory(samlp.Response,
                assertion=assertion2,
                id="22233",
                signature=sigver.pre_signature_part("22233", self.sec.my_cert))

        to_sign = [(class_name(assertion2), assertion2.id),
                    (class_name(response), response.id)]

        s_response = sigver.signed_instance_factory(response, self.sec, to_sign)

        assert s_response is not None
        response2 = response_from_string(s_response)

        sass = response2.assertion[0]
        assert _eq(sass.keyswv(), ['attribute_statement', 'issue_instant',
                                'version', 'signature', 'id'])
        assert sass.version == "2.0"
        assert sass.id == "11122"

        item = self.sec.check_signature(response2, class_name(response),
                                        s_response)

        assert isinstance(item, samlp.Response)

def test_xbox_non_ascii_ava():
    conf = config.SPConfig()
    conf.load_file("server_conf")
    md = MetadataStore([saml, samlp], None, conf)
    md.load("local", IDP_EXAMPLE)

    conf.metadata = md
    conf.only_use_keys_in_metadata = False
    sec = sigver.security_context(conf)

    assertion = factory(
        saml.Assertion, version="2.0", id="11111",
        issue_instant="2009-10-30T13:20:28Z",
        signature=sigver.pre_signature_part("11111", sec.my_cert, 1),
        attribute_statement=do_attribute_statement(
            {
                ("", "", "surName"): ("Föö", ""),
                ("", "", "givenName"): ("Bär", ""),
            }
        )
    )

    sigass = sec.sign_statement(
        assertion,
        class_name(assertion),
        key_file=PRIV_KEY,
        node_id=assertion.id,
    )

    _ass0 = saml.assertion_from_string(sigass)
    encrypted_assertion = EncryptedAssertion()
    encrypted_assertion.add_extension_element(_ass0)

    _, pre = make_temp(
        str(pre_encryption_part()).encode('utf-8'), decode=False
    )
    enctext = sec.crypto.encrypt(
        str(encrypted_assertion),
        conf.cert_file,
        pre,
        "des-192",
        '/*[local-name()="EncryptedAssertion"]/*[local-name()="Assertion"]',
    )

    decr_text = sec.decrypt(enctext, key_file=PRIV_KEY)
    _seass = saml.encrypted_assertion_from_string(decr_text)
    assertions = []
    assers = extension_elements_to_elements(
        _seass.extension_elements, [saml, samlp]
    )

    for ass in assers:
        _txt = sec.verify_signature(
            str(ass), PUB_KEY, node_name=class_name(assertion)
        )
        if _txt:
            assertions.append(ass)

    assert assertions
    print(assertions)

    def correctly_signed_response(self, decoded_xml, must=False, origdoc=None):
        """ Check if a instance is correctly signed, if we have metadata for
        the IdP that sent the info use that, if not use the key that are in 
        the message if any.
        
        :param decoded_xml: The SAML message as a XML string
        :param must: Whether there must be a signature
        :return: None if the signature can not be verified otherwise an instance
        """
        
        response = samlp.response_from_string(decoded_xml)
        if not response:
            raise TypeError("Not a Response")

        if response.signature:
            self._check_signature(decoded_xml, response, class_name(response),
                                  origdoc)

        if response.assertion:
            # Try to find the signing cert in the assertion
            for assertion in response.assertion:
                if not assertion.signature:
                    logger.debug("unsigned")
                    if must:
                        raise SignatureError("Signature missing")
                    continue
                else:
                    logger.debug("signed")

                try:
                    self._check_signature(decoded_xml, assertion,
                                            class_name(assertion), origdoc)
                except Exception, exc:
                    logger.error("correctly_signed_response: %s" % exc)
                    raise

    def test_sign_response(self):
        response = factory(samlp.Response,
                           assertion=self._assertion,
                           id="22222",
                           signature=sigver.pre_signature_part("22222",
                                                               self.sec
                                                               .my_cert))

        to_sign = [(class_name(self._assertion), self._assertion.id),
                   (class_name(response), response.id)]
        s_response = sigver.signed_instance_factory(response, self.sec, to_sign)

        assert s_response is not None
        print(s_response)
        response = response_from_string(s_response)
        sass = response.assertion[0]

        print(sass)
        assert _eq(sass.keyswv(), ['attribute_statement', 'issue_instant',
                                   'version', 'signature', 'id'])
        assert sass.version == "2.0"
        assert sass.id == "11111"

        item = self.sec.check_signature(response, class_name(response),
                                        s_response)
        assert isinstance(item, samlp.Response)
        assert item.id == "22222"

    def test_multiple_signatures_response(self):
        response = factory(
            samlp.Response,
            assertion=self._assertion,
            id="22222",
            signature=sigver.pre_signature_part("22222", self.sec.my_cert),
        )

        # order is important, we can't validate if the signatures are made
        # in the reverse order
        to_sign = [(self._assertion, self._assertion.id, ""), (response, response.id, "")]

        s_response = self.sec.multiple_signatures("%s" % response, to_sign)
        assert s_response is not None
        response = response_from_string(s_response)

        item = self.sec.check_signature(response, class_name(response), s_response, must=True)
        assert item == response
        assert item.id == "22222"

        s_assertion = item.assertion[0]
        assert isinstance(s_assertion, saml.Assertion)
        # make sure the assertion was modified when we supposedly signed it
        assert s_assertion != self._assertion

        ci = "".join(sigver.cert_from_instance(s_assertion)[0].split())
        assert ci == self.sec.my_cert

        res = self.sec.check_signature(s_assertion, class_name(s_assertion), s_response, must=True)
        assert res == s_assertion
        assert s_assertion.id == "11111"
        assert s_assertion.version == "2.0"
        assert _eq(s_assertion.keyswv(), ["attribute_statement", "issue_instant", "version", "signature", "id"])

    def test_sign_verify_assertion_with_cert_from_instance(self):
        assertion = factory(saml.Assertion,
                            version="2.0",
                            id="11100",
                            issue_instant="2009-10-30T13:20:28Z",
                            signature=sigver.pre_signature_part("11100",
                                                                self.sec
                                                                .my_cert),
                            attribute_statement=do_attribute_statement({
                                ("", "", "surName"): ("Fox", ""),
                                ("", "", "givenName"): ("Bear", ""),
                            })
        )

        to_sign = [(class_name(assertion), assertion.id)]
        s_assertion = sigver.signed_instance_factory(assertion, self.sec,
                                                     to_sign)
        print(s_assertion)
        ass = assertion_from_string(s_assertion)
        ci = "".join(sigver.cert_from_instance(ass)[0].split())
        assert ci == self.sec.my_cert

        res = self.sec.verify_signature(s_assertion,
                                        node_name=class_name(ass))
        assert res

        res = self.sec._check_signature(s_assertion, ass, class_name(ass))

        assert res

def test_create_class_from_xml_string_nameid():
    kl = create_class_from_xml_string(NameID, ITEMS[NameID][0])
    assert kl != None
    assert kl.format == "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress"
    assert kl.sp_provided_id == "sp provided id"
    assert kl.text.strip() == "[email protected]"
    assert _eq(kl.keyswv(), ['sp_provided_id', 'format', 'text'])
    assert class_name(kl) == "urn:oasis:names:tc:SAML:2.0:assertion:NameID"
    assert _eq(kl.keys(), ['sp_provided_id', 'sp_name_qualifier',
                           'name_qualifier', 'format', 'text'])

    kl = create_class_from_xml_string(NameID, ITEMS[NameID][1])
    assert kl != None
    assert kl.format == "urn:oasis:names:tc:SAML:2.0:nameid-format:transient"
    assert kl.sp_name_qualifier == "https://foo.example.com/sp"
    assert kl.text.strip() == "_1632879f09d08ea5ede2dc667cbed7e429ebc4335c"
    assert _eq(kl.keyswv(), ['sp_name_qualifier', 'format', 'text'])
    assert class_name(kl) == "urn:oasis:names:tc:SAML:2.0:assertion:NameID"

    kl = create_class_from_xml_string(NameID, ITEMS[NameID][2])
    assert kl != None
    assert kl.format == "urn:oasis:names:tc:SAML:2.0:nameid-format:persistent"
    assert kl.name_qualifier == "http://authentic.example.com/saml/metadata"
    assert kl.sp_name_qualifier == "http://auth.example.com/saml/metadata"
    assert kl.text.strip() == "test"
    assert _eq(kl.keyswv(), ['sp_name_qualifier', 'format', 'name_qualifier',
                             'text'])
    assert class_name(kl) == "urn:oasis:names:tc:SAML:2.0:assertion:NameID"

    def test_sign_verify_with_cert_from_instance(self):
        response = factory(samlp.Response,
                           assertion=self._assertion,
                           id="22222",
                           signature=sigver.pre_signature_part("22222",
                                                               self.sec
                                                               .my_cert))

        to_sign = [(class_name(self._assertion), self._assertion.id),
                   (class_name(response), response.id)]

        s_response = sigver.signed_instance_factory(response, self.sec, to_sign)

        response2 = response_from_string(s_response)

        ci = "".join(sigver.cert_from_instance(response2)[0].split())

        assert ci == self.sec.my_cert

        res = self.sec.verify_signature(s_response,
                                        node_name=class_name(samlp.Response()))

        assert res

        res = self.sec._check_signature(s_response, response2,
                                        class_name(response2), s_response)
        assert res == response2

def test_xbox():
    conf = config.SPConfig()
    conf.load_file("server_conf")
    md = MetadataStore([saml, samlp], None, conf)
    md.load("local", full_path("idp_example.xml"))

    conf.metadata = md
    conf.only_use_keys_in_metadata = False
    sec = sigver.security_context(conf)

    assertion = factory(
        saml.Assertion,
        version="2.0",
        id="11111",
        issue_instant="2009-10-30T13:20:28Z",
        signature=sigver.pre_signature_part("11111", sec.my_cert, 1),
        attribute_statement=do_attribute_statement(
            {("", "", "surName"): ("Foo", ""), ("", "", "givenName"): ("Bar", "")}
        ),
    )

    sigass = sec.sign_statement(assertion, class_name(assertion), key_file=full_path("test.key"), node_id=assertion.id)

    _ass0 = saml.assertion_from_string(sigass)

    encrypted_assertion = EncryptedAssertion()
    encrypted_assertion.add_extension_element(_ass0)

    _, pre = make_temp(str(pre_encryption_part()).encode("utf-8"), decode=False)
    enctext = sec.crypto.encrypt(
        str(encrypted_assertion),
        conf.cert_file,
        pre,
        "des-192",
        '/*[local-name()="EncryptedAssertion"]/*[local-name()="Assertion"]',
    )

    decr_text = sec.decrypt(enctext)
    _seass = saml.encrypted_assertion_from_string(decr_text)
    assertions = []
    assers = extension_elements_to_elements(_seass.extension_elements, [saml, samlp])

    sign_cert_file = full_path("test.pem")

    for ass in assers:
        _ass = "%s" % ass
        # _ass = _ass.replace('xsi:nil="true" ', '')
        # assert sigass == _ass
        _txt = sec.verify_signature(_ass, sign_cert_file, node_name=class_name(assertion))
        if _txt:
            assertions.append(ass)

    print(assertions)

    def sign(self, msg, mid=None, to_sign=None):
        if msg.signature is None:
            msg.signature = pre_signature_part(msg.id, self.sec.my_cert, 1)

        if mid is None:
            mid = msg.id

        try:
            to_sign.append([(class_name(msg), mid)])
        except AttributeError:
            to_sign = [(class_name(msg), mid)]

        logger.info("REQUEST: %s" % msg)

        return signed_instance_factory(msg, self.sec, to_sign)

    def test_sign_auth_request_0(self):
        #print self.client.config
        
        ar_str = "%s" % self.client.authn_request("id1",
                                        "http://www.example.com/sso",
                                        "http://www.example.org/service",
                                        "urn:mace:example.org:saml:sp",
                                        "My Name", sign=True)
                                    
        ar = samlp.authn_request_from_string(ar_str)

        assert ar
        assert ar.signature
        assert ar.signature.signature_value
        signed_info = ar.signature.signed_info
        #print signed_info
        assert len(signed_info.reference) == 1
        assert signed_info.reference[0].uri == "#id1"
        assert signed_info.reference[0].digest_value
        print "------------------------------------------------"
        try:
            assert self.client.sec.correctly_signed_authn_request(ar_str,
                    self.client.config.xmlsec_binary,
                    self.client.config.metadata)
        except Exception: # missing certificate
            self.client.sec.verify_signature(ar_str, node_name=class_name(ar))

    def use_soap(self, request, destination="", soap_headers=None, sign=False,
                 **kwargs):
        """
        Construct the necessary information for using SOAP+POST

        :param request:
        :param destination:
        :param soap_headers:
        :param sign:
        :return: dictionary
        """
        headers = [("content-type", "application/soap+xml")]

        soap_message = make_soap_enveloped_saml_thingy(request, soap_headers)

        logger.debug("SOAP message: %s", soap_message)

        if sign and self.sec:
            _signed = self.sec.sign_statement(soap_message,
                                              class_name=class_name(request),
                                              node_id=request.id)
            soap_message = _signed

        return {"url": destination, "method": "POST",
                "data": soap_message, "headers": headers}

    def test_sign_auth_request_0(self):
        #print self.client.config
        
        req_id, a_req = self.client.create_authn_request(
            "http://www.example.com/sso", sign=True, message_id="id1")

        if isinstance(a_req, bytes):
            ar_str = a_req
        else:
            ar_str = a_req.to_string()
        ar = samlp.authn_request_from_string(ar_str)

        assert ar
        assert ar.signature
        assert ar.signature.signature_value
        signed_info = ar.signature.signed_info
        #print signed_info
        assert len(signed_info.reference) == 1
        assert signed_info.reference[0].uri == "#id1"
        assert signed_info.reference[0].digest_value
        print("------------------------------------------------")
        try:
            assert self.client.sec.correctly_signed_authn_request(
                ar_str, self.client.config.xmlsec_binary,
                self.client.config.metadata)
        except Exception:  # missing certificate
            self.client.sec.verify_signature(ar_str, node_name=class_name(ar))

    def test_SAML_sign_with_pkcs11(self):
        """
        Test signing a SAML assertion using PKCS#11 and then verifying it.
        """
        os.environ['SOFTHSM_CONF'] = self.softhsm_conf

        ass = self._assertion
        print ass
        sign_ass = self.sec.sign_assertion("%s" % ass, node_id=ass.id)
        #print sign_ass
        sass = saml.assertion_from_string(sign_ass)
        #print sass
        assert _eq(sass.keyswv(), ['attribute_statement', 'issue_instant',
                                   'version', 'signature', 'id'])
        assert sass.version == "2.0"
        assert sass.id == "11111"
        assert time_util.str_to_time(sass.issue_instant)

        print "Crypto version : %s" % (self.sec.crypto.version())

        item = self.sec.check_signature(sass, class_name(sass), sign_ass)

        assert isinstance(item, saml.Assertion)

        print "Test PASSED"

def test_create_class_from_xml_string_subject_locality():
    kl = create_class_from_xml_string(SubjectLocality, ITEMS[SubjectLocality])
    assert kl != None
    assert _eq(kl.keyswv(), ['address', "dns_name"])
    assert kl.address == "127.0.0.1"
    assert kl.dns_name == "localhost"
    assert class_name(kl) == "urn:oasis:names:tc:SAML:2.0:assertion:SubjectLocality"

    def do_authz_decision_query(self, entityid, assertion=None, log=None, sign=False):

        authz_decision_query = self.authz_decision_query(entityid, assertion)

        for destination in self.config.authz_services(entityid):
            to_sign = []
            if sign:
                authz_decision_query.signature = pre_signature_part(authz_decision_query.id, self.sec.my_cert, 1)
                to_sign.append((class_name(authz_decision_query), authz_decision_query.id))

                authz_decision_query = signed_instance_factory(authz_decision_query, self.sec, to_sign)

            response = send_using_soap(
                authz_decision_query,
                destination,
                self.config.key_file,
                self.config.cert_file,
                log=log,
                ca_certs=self.config.ca_certs,
            )
            if response:
                if log:
                    log.info("Verifying response")
                response = self.authz_decision_query_response(response, log)

            if response:
                # not_done.remove(entity_id)
                if log:
                    log.info("OK response from %s" % destination)
                return response
            else:
                if log:
                    log.info("NOT OK response from %s" % destination)

        return None

    def create_attribute_response(self, identity, in_response_to, destination,
                                  sp_entity_id, userid="", name_id=None,
                                  status=None, issuer=None,
                                  sign_assertion=False, sign_response=False,
                                  attributes=None):
        """ Create an attribute assertion response.
        
        :param identity: A dictionary with attributes and values that are
            expected to be the bases for the assertion in the response.
        :param in_response_to: The session identifier of the request
        :param destination: The URL which should receive the response
        :param sp_entity_id: The entity identifier of the SP
        :param userid: A identifier of the user
        :param name_id: The identifier of the subject
        :param status: The status of the response
        :param issuer: The issuer of the response
        :param sign_assertion: Whether the assertion should be signed or not
        :param sign_response: Whether the whole response should be signed
        :param attributes:
        :return: A response instance
        """
        if not name_id and userid:
            try:
                name_id = self.ident.construct_nameid(userid,
                                                      self.config.policy,
                                                      sp_entity_id)
                logger.warning("Unspecified NameID format")
            except Exception:
                pass

        to_sign = []
        args = {}
        if identity:
            _issuer = self._issuer(issuer)
            ast = Assertion(identity)
            policy = self.config.getattr("policy", "aa")
            if policy:
                ast.apply_policy(sp_entity_id, policy)
            else:
                policy = Policy()

            if attributes:
                restr = restriction_from_attribute_spec(attributes)
                ast = filter_attribute_value_assertions(ast)

            assertion = ast.construct(sp_entity_id, in_response_to,
                                      destination, name_id,
                                      self.config.attribute_converters,
                                      policy, issuer=_issuer)

            if sign_assertion:
                assertion.signature = pre_signature_part(assertion.id,
                                                         self.sec.my_cert, 1)
                # Just the assertion or the response and the assertion ?
                to_sign = [(class_name(assertion), assertion.id)]

            args["assertion"] = assertion

        return self._response(in_response_to, destination, status, issuer,
                              sign_response, to_sign, **args)

    def decrypt_assertions(self, encrypted_assertions, decr_txt, issuer=None, verified=False):
        """ Moves the decrypted assertion from the encrypted assertion to a
        list.

        :param encrypted_assertions: A list of encrypted assertions.
        :param decr_txt: The string representation containing the decrypted
        data. Used when verifying signatures.
        :param issuer: The issuer of the response.
        :param verified: If True do not verify signatures, otherwise verify
        the signature if it exists.
        :return: A list of decrypted assertions.
        """
        res = []
        for encrypted_assertion in encrypted_assertions:
            if encrypted_assertion.extension_elements:
                assertions = extension_elements_to_elements(encrypted_assertion.extension_elements, [saml, samlp])
                for assertion in assertions:
                    if assertion.signature and not verified:
                        if not self.sec.check_signature(
                            assertion, origdoc=decr_txt, node_name=class_name(assertion), issuer=issuer
                        ):
                            logger.error("Failed to verify signature on '%s'", assertion)
                            raise SignatureError()
                    res.append(assertion)
        return res

    def correctly_signed_message(self, decoded_xml, msgtype, must=False,
                                 origdoc=None):
        """Check if a request is correctly signed, if we have metadata for
        the entity that sent the info use that, if not use the key that are in
        the message if any.

        :param decoded_xml: The SAML message as a XML string
        :param msgtype:
        :param must: Whether there must be a signature
        :param origdoc:
        :return:
        """

        try:
            _func = getattr(samlp, "%s_from_string" % msgtype)
        except AttributeError:
            _func = getattr(saml, "%s_from_string" % msgtype)

        msg = _func(decoded_xml)
        if not msg:
            raise TypeError("Not a %s" % msgtype)

        if not msg.signature:
            if must:
                raise SignatureError("Missing must signature")
            else:
                return msg

        return self._check_signature(decoded_xml, msg, class_name(msg),
                                     origdoc, must=must)