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_create_class_from_xml_string_xxe():
    xml = """<?xml version="1.0"?>
    <!DOCTYPE lolz [
    <!ENTITY lol "lol">
    <!ELEMENT lolz (#PCDATA)>
    <!ENTITY lol1 "&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;">
    ]>
    <lolz>&lol1;</lolz>
    """
    with raises(EntitiesForbidden) as err:
        create_class_from_xml_string(NameID, xml)

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 test_subject_confirmation_with_extension():
    kl = create_class_from_xml_string(SubjectConfirmation,
                                      SUBJECT_CONFIRMATION_WITH_MEMBER_EXTENSION)
    assert kl != None
    print kl.__dict__
    assert kl.extension_attributes == {}
    assert kl.method == "urn:oasis:names:tc:SAML:2.0:cm:bearer"
    name_id = kl.name_id
    assert _eq(name_id.keyswv(), ['format', 'name_qualifier', 'text'])
    assert name_id.format == "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress"
    assert name_id.name_qualifier == "http://authentic.example.com/saml/metadata"
    assert name_id.text.strip() == "[email protected]"
    subject_confirmation_data = kl.subject_confirmation_data
    assert _eq(subject_confirmation_data.keyswv(), ['not_on_or_after',
                                                    'recipient',
                                                    'in_response_to'])
    assert subject_confirmation_data.recipient == \
           "http://auth.example.com/saml/proxySingleSignOnRedirect"
    assert subject_confirmation_data.not_on_or_after == "2010-02-17T17:02:38Z"
    assert subject_confirmation_data.in_response_to == \
           "_59B3A01B03334032C31E434C63F89E3E"
    assert len(kl.extension_elements) == 1
    ee = kl.extension_elements[0]
    assert ee.tag == "Trustlevel"
    assert ee.namespace == "urn:mace:example.com:saml:assertion"
    assert ee.text.strip() == "Excellent"

def test_to_fro_string_1():
    kl = create_class_from_xml_string(SubjectConfirmation,
                                      SUBJECT_CONFIRMATION_WITH_MEMBER_EXTENSION)
    txt = kl.to_string()
    cpy = create_class_from_xml_string(SubjectConfirmation, txt)

    print kl.__dict__
    print cpy.__dict__

    assert kl.text.strip() == cpy.text.strip()
    assert _eq(kl.keyswv(), cpy.keyswv())
    assert len(kl.extension_elements) == len(cpy.extension_elements)
    klee = kl.extension_elements[0]
    cpyee = cpy.extension_elements[0]
    assert klee.text.strip() == cpyee.text.strip()
    assert klee.tag == cpyee.tag
    assert klee.namespace == cpyee.namespace

def _sign_assertion(assertion):
    """Sign a SAML assertion.

    This method utilizes ``xmlsec1`` binary and signs SAML assertions in a
    separate process. ``xmlsec1`` cannot read input data from stdin so the
    prepared assertion needs to be serialized and stored in a temporary
    file. This file will be deleted immediately after ``xmlsec1`` returns.
    The signed assertion is redirected to a standard output and read using
    subprocess.PIPE redirection. A ``saml.Assertion`` class is created
    from the signed string again and returned.

    Parameters that are required in the CONF::
    * xmlsec_binary
    * private key file path
    * public key file path
    :return: XML <Assertion> object

    """
    xmlsec_binary = CONF.saml.xmlsec1_binary
    idp_private_key = CONF.saml.keyfile
    idp_public_key = CONF.saml.certfile

    # xmlsec1 --sign --privkey-pem privkey,cert --id-attr:ID <tag> <file>
    certificates = '%(idp_private_key)s,%(idp_public_key)s' % {
        'idp_public_key': idp_public_key,
        'idp_private_key': idp_private_key
    }

    command_list = [xmlsec_binary, '--sign', '--privkey-pem', certificates,
                    '--id-attr:ID', 'Assertion']

    try:
        # NOTE(gyee): need to make the namespace prefixes explicit so
        # they won't get reassigned when we wrap the assertion into
        # SAML2 response
        file_path = fileutils.write_to_tempfile(assertion.to_string(
            nspair={'saml': saml2.NAMESPACE,
                    'xmldsig': xmldsig.NAMESPACE}))
        command_list.append(file_path)
        process = subprocess.Popen(command_list,
                                   stdin=subprocess.PIPE,
                                   stdout=subprocess.PIPE,
                                   stderr=subprocess.PIPE,
                                   close_fds=True)
        stdout, stderr = process.communicate()
        retcode = process.poll()
        if retcode:
            msg = _LE('Error when signing assertion, reason: %(reason)s')
            msg = msg % {'reason': stderr}
            LOG.error(msg)
            raise exception.SAMLSigningError(reason=stderr)
    finally:
        try:
            os.remove(file_path)
        except OSError:
            pass

    return saml2.create_class_from_xml_string(saml.Assertion, stdout)

def test_create_class_from_xml_string_subject_confirmation_data():
    kl = create_class_from_xml_string(SubjectConfirmationData,
                                      ITEMS[SubjectConfirmationData])
    assert kl != None
    assert _eq(kl.keyswv(), ['in_response_to', 'not_on_or_after',
                             'not_before', 'recipient'])
    assert kl.in_response_to == "_1683146e27983964fbe7bf8f08961108d166a652e5"
    assert kl.not_on_or_after == "2010-02-18T13:52:13.959Z"
    assert kl.not_before == "2010-01-16T12:00:00Z"
    assert kl.recipient == "http://192.168.0.10/saml/sp"
    assert class_name(kl) == \
           "urn:oasis:names:tc:SAML:2.0:assertion:SubjectConfirmationData"

def signed_instance_factory(instance, seccont, elements_to_sign=None):
    if elements_to_sign:
        signed_xml = "%s" % instance
        for (node_name, nodeid) in elements_to_sign:
            signed_xml = seccont.sign_statement_using_xmlsec(signed_xml,
                                    klass_namn=node_name, nodeid=nodeid)

        #print "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
        #print "%s" % signed_xml
        #print "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
        return create_class_from_xml_string(instance.__class__, signed_xml)
    else:
        return instance

def test_nameid_with_extension():
    kl = create_class_from_xml_string(NameID, NAMEID_WITH_ATTRIBUTE_EXTENSION)
    assert kl != None
    print kl.__dict__
    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',
                             'extension_attributes', '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'])
    assert kl.extension_attributes == {
        '{urn:mace:example.com:saml:assertion}Foo': 'BAR'}

def test_attribute_element_to_extension_element():
    attr = create_class_from_xml_string(Attribute, saml2_data.TEST_ATTRIBUTE)
    ee = saml2.element_to_extension_element(attr)
    print ee.__dict__
    assert ee.tag == "Attribute"
    assert ee.namespace == 'urn:oasis:names:tc:SAML:2.0:assertion'
    assert _eq(ee.attributes.keys(), ['FriendlyName', 'Name', 'NameFormat'])
    assert ee.attributes["FriendlyName"] == 'test attribute'
    assert ee.attributes["Name"] == "testAttribute"
    assert ee.attributes["NameFormat"] == \
           'urn:oasis:names:tc:SAML:2.0:attrname-format:unspecified'
    assert len(ee.children) == 2
    for child in ee.children:
        # children are also extension element instances
        assert child.namespace == 'urn:oasis:names:tc:SAML:2.0:assertion'
        assert child.tag == "AttributeValue"

def _sign_assertion(assertion):
    """Sign a SAML assertion.

    This method utilizes ``xmlsec1`` binary and signs SAML assertions in a
    separate process. ``xmlsec1`` cannot read input data from stdin so the
    prepared assertion needs to be serialized and stored in a temporary
    file. This file will be deleted immediately after ``xmlsec1`` returns.
    The signed assertion is redirected to a standard output and read using
    subprocess.PIPE redirection. A ``saml.Assertion`` class is created
    from the signed string again and returned.

    Parameters that are required in the CONF::
    * xmlsec_binary
    * private key file path
    * public key file path
    :return: XML <Assertion> object

    """
    xmlsec_binary = CONF.saml.xmlsec1_binary
    idp_private_key = CONF.saml.keyfile
    idp_public_key = CONF.saml.certfile

    # xmlsec1 --sign --privkey-pem privkey,cert --id-attr:ID <tag> <file>
    certificates = '%(idp_private_key)s,%(idp_public_key)s' % {
        'idp_public_key': idp_public_key,
        'idp_private_key': idp_private_key
    }

    command_list = [xmlsec_binary, '--sign', '--privkey-pem', certificates,
                    '--id-attr:ID', 'Assertion']

    try:
        file_path = fileutils.write_to_tempfile(assertion.to_string())
        command_list.append(file_path)
        stdout = subprocess.check_output(command_list)
    except Exception as e:
        msg = _LE('Error when signing assertion, reason: %(reason)s')
        msg = msg % {'reason': e}
        LOG.error(msg)
        raise exception.SAMLSigningError(reason=e)
    finally:
        try:
            os.remove(file_path)
        except OSError:
            pass

    return saml2.create_class_from_xml_string(saml.Assertion, stdout)

def test_create_class_from_xml_string_subject_confirmation():
    kl = create_class_from_xml_string(SubjectConfirmation, 
                                        ITEMS[SubjectConfirmation])
    assert kl != None
    assert _eq(kl.keyswv(), ['method', 'name_id',
                                'subject_confirmation_data'])
    assert kl.method == "urn:oasis:names:tc:SAML:2.0:cm:bearer"
    name_id = kl.name_id
    assert _eq(name_id.keyswv(), ['format', 'name_qualifier', 'text'])
    assert name_id.format == "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress"
    assert name_id.name_qualifier == "http://authentic.example.com/saml/metadata"
    assert name_id.text.strip() == "[email protected]"
    subject_confirmation_data = kl.subject_confirmation_data
    assert _eq(subject_confirmation_data.keyswv(), ['not_on_or_after', 
                            'recipient', 'in_response_to'])
    assert subject_confirmation_data.recipient == \
                    "http://auth.example.com/saml/proxySingleSignOnRedirect"
    assert subject_confirmation_data.not_on_or_after == "2010-02-17T17:02:38Z"
    assert subject_confirmation_data.in_response_to == \
                    "_59B3A01B03334032C31E434C63F89E3E"
    assert class_name(kl) == \
                "urn:oasis:names:tc:SAML:2.0:assertion:SubjectConfirmation"

def response_type__from_string(xml_string):
    return saml2.create_class_from_xml_string(ResponseType_, xml_string)

def request_type__from_string(xml_string):
    return saml2.create_class_from_xml_string(RequestType_, xml_string)

def _sign_assertion(assertion):
    """Sign a SAML assertion.

    This method utilizes ``xmlsec1`` binary and signs SAML assertions in a
    separate process. ``xmlsec1`` cannot read input data from stdin so the
    prepared assertion needs to be serialized and stored in a temporary file.
    This file will be deleted immediately after ``xmlsec1`` returns. The signed
    assertion is redirected to a standard output and read using
    ``subprocess.PIPE`` redirection. A ``saml.Assertion`` class is created from
    the signed string again and returned.

    Parameters that are required in the CONF::
    * xmlsec_binary
    * private key file path
    * public key file path
    :returns: XML <Assertion> object

    """
    # Ensure that the configured certificate paths do not contain any commas,
    # before we string format a comma in between them and cause xmlsec1 to
    # explode like a thousand fiery supernovas made entirely of unsigned SAML.
    for option in ('keyfile', 'certfile'):
        if ',' in getattr(CONF.saml, option, ''):
            raise exception.UnexpectedError(
                'The configuration value in `keystone.conf [saml] %s` cannot '
                'contain a comma (`,`). Please fix your configuration.' %
                option)

    # xmlsec1 --sign --privkey-pem privkey,cert --id-attr:ID <tag> <file>
    certificates = '%(idp_private_key)s,%(idp_public_key)s' % {
        'idp_public_key': CONF.saml.certfile,
        'idp_private_key': CONF.saml.keyfile,
    }

    # Verify that the binary used to create the assertion actually exists on
    # the system. If it doesn't, log a warning for operators to go and install
    # it. Requests for assertions will fail with HTTP 500s until the package is
    # installed, so providing something useful in the logs is about the best we
    # can do.
    _verify_assertion_binary_is_installed()

    command_list = [
        CONF.saml.xmlsec1_binary, '--sign', '--privkey-pem', certificates,
        '--id-attr:ID', 'Assertion']

    file_path = None
    try:
        # NOTE(gyee): need to make the namespace prefixes explicit so
        # they won't get reassigned when we wrap the assertion into
        # SAML2 response
        file_path = fileutils.write_to_tempfile(assertion.to_string(
            nspair={'saml': saml2.NAMESPACE,
                    'xmldsig': xmldsig.NAMESPACE}))
        command_list.append(file_path)
        stdout = subprocess.check_output(command_list,  # nosec : The contents
                                         # of the command list are coming from
                                         # a trusted source because the
                                         # executable and arguments all either
                                         # come from the config file or are
                                         # hardcoded. The command list is
                                         # initialized earlier in this function
                                         # to a list and it's still a list at
                                         # this point in the function. There is
                                         # no opportunity for an attacker to
                                         # attempt command injection via string
                                         # parsing.
                                         stderr=subprocess.STDOUT)
    except Exception as e:
        msg = 'Error when signing assertion, reason: %(reason)s%(output)s'
        LOG.error(msg,
                  {'reason': e,
                   'output': ' ' + e.output if hasattr(e, 'output') else ''})
        raise exception.SAMLSigningError(reason=e)
    finally:
        try:
            if file_path:
                os.remove(file_path)
        except OSError:  # nosec
            # The file is already gone, good.
            pass

    return saml2.create_class_from_xml_string(saml.Assertion, stdout)

def applies_to_from_string(xml_string):
    return saml2.create_class_from_xml_string(AppliesTo, xml_string)

def exactly_one_from_string(xml_string):
    return saml2.create_class_from_xml_string(ExactlyOne, xml_string)

def policy_from_string(xml_string):
    return saml2.create_class_from_xml_string(Policy, xml_string)

def request_initiator_from_string(xml_string):
    return saml2.create_class_from_xml_string(RequestInitiator, xml_string)

def request_from_string(xml_string):
    return saml2.create_class_from_xml_string(Request, xml_string)