def load_metadata(self, metadata_conf):
        """ Loads metadata into an internal structure """

        xmlsec_binary = self.xmlsec_binary
        acs = self.attribute_converters

        if xmlsec_binary is None:
            raise Exception("Missing xmlsec1 specification")
        if acs is None:
            raise Exception("Missing attribute converter specification")

        try:
            ca_certs = self.ca_certs
        except:
            ca_certs = None
        try:
            disable_validation = self.disable_ssl_certificate_validation
        except:
            disable_validation = False

        mds = MetadataStore(ONTS.values(), acs, xmlsec_binary, ca_certs,
                            disable_ssl_certificate_validation=disable_validation)

        mds.imp(metadata_conf)

        return mds

def test_metadata_file():
    sec_config.xmlsec_binary = sigver.get_xmlsec_binary(["/opt/local/bin"])
    mds = MetadataStore(list(ONTS.values()), ATTRCONV, sec_config, disable_ssl_certificate_validation=True)

    mds.imp(METADATACONF["8"])
    print((len(list(mds.keys()))))
    assert len(list(mds.keys())) == 560

    def load_metadata(self, metadata_conf):
        """ Loads metadata into an internal structure """

        acs = self.attribute_converters

        if acs is None:
            raise ConfigurationError(
                "Missing attribute converter specification")

        try:
            ca_certs = self.ca_certs
        except:
            ca_certs = None
        try:
            disable_validation = self.disable_ssl_certificate_validation
        except:
            disable_validation = False

        mds = MetadataStore(
            ONTS.values(), acs, self, ca_certs,
            disable_ssl_certificate_validation=disable_validation)

        mds.imp(metadata_conf)

        return mds

def test_xmlsec_err_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", ""), })
    )

    with raises(XmlsecError):
        sec.sign_statement(
            assertion,
            class_name(assertion),
            key_file=INVALID_KEY,
            node_id=assertion.id,
        )

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 test_get_certs_from_metadata():
    mds = MetadataStore(ONTS.values(), ATTRCONV, None)
    mds.imp(METADATACONF["11"])
    certs1 = mds.certs("http://xenosmilus.umdc.umu.se/simplesaml/saml2/idp/metadata.php", "any")
    certs2 = mds.certs("http://xenosmilus.umdc.umu.se/simplesaml/saml2/idp/metadata.php", "idpsso")

    assert certs1[0] == certs2[0] == TEST_CERT

def test_xmlsec_err():
    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", ""), })
    )

    try:
        sec.sign_statement(assertion, class_name(assertion),
                           key_file=full_path("tes.key"),
                           node_id=assertion.id)
    except (XmlsecError, SigverError) as err:  # should throw an exception
        pass
    else:
        assert False

def test_extension():
    mds = MetadataStore(ATTRCONV, None)
    # use ordered dict to force expected entity to be last
    metadata = OrderedDict()
    metadata["1"] = {"entity1": {}}
    metadata["2"] = {"entity2": {"idpsso_descriptor": [{"extensions": {"extension_elements": [{"__class__": "test"}]}}]}}
    mds.metadata = metadata
    assert mds.extension("entity2", "idpsso_descriptor", "test")

def test_load_local_dir():
    sec_config.xmlsec_binary = sigver.get_xmlsec_binary(["/opt/local/bin"])
    mds = MetadataStore(ONTS.values(), ATTRCONV, sec_config, disable_ssl_certificate_validation=True)

    mds.imp(METADATACONF["9"])
    print mds
    assert len(mds) == 3  # Three sources
    assert len(mds.keys()) == 4  # number of idps

    def __init__(self, idp_conf, logger, conf, publicKey, privateKey, metadataList):
        """
        Constructor.
        Initiates the class.
        :param logger: Logger to be used when something needs to be logged.
        :param conf: idp_proxy_conf see IdpProxy/conig/idp_proxy_conf.example.py
        :param key: A RSA key to be used for encryption.
        :param metadataList: A list of metadata files.
            [{"local": ["swamid-1.0.xml"]}, {"local": ["sp.xml"]}]
        :raise:
        """
        if (logger is None) or (conf is None) or (publicKey is None)or (privateKey is None):
            raise ValueError(
                "A new instance must include a value for logger, conf and key.")
        #Public key to be used for encryption.
        self.publicKey = publicKey
        self.privateKey = privateKey
        #Used for presentation of mako files.
        self.lookup = TemplateLookup(
            directories=[MetadataGeneration.CONST_STATIC_MAKO + 'templates',
                         MetadataGeneration.CONST_STATIC_MAKO + 'htdocs'],
            module_directory='modules',
            input_encoding='utf-8',
            output_encoding='utf-8')
        #The logger.
        self.logger = logger
        #A list of all social services used by this IdPproxy.
        self.socialServiceKeyList = []
        #A list of all service providers used by this sp.
        self.spKeyList = []
        for key in conf:
            self.socialServiceKeyList.append(conf[key]["name"])

        try:
            xmlsec_path = get_xmlsec_binary(["/opt/local/bin"])
        except:
            try:
                xmlsec_path = get_xmlsec_binary(["/usr/local/bin"])
            except:
                self.logger.info('Xmlsec must be installed! Tries /usr/bin/xmlsec1.')
                xmlsec_path = '/usr/bin/xmlsec1'

        self.xmlsec_path = xmlsec_path

        config = Config()
        config.disable_ssl_certificate_validation = True
        config.key_file = idp_conf["key_file"]
        config.cert_file = idp_conf["cert_file"]
        config.xmlsec_binary = idp_conf["xmlsec_binary"]
        config.debug = idp_conf["debug"]

        for metadata in metadataList:
            mds = MetadataStore(MetadataGeneration.CONST_ONTS.values(),
                                MetadataGeneration.CONST_ATTRCONV, config)
            mds.imp(metadata)
            for entityId in mds.keys():
                self.spKeyList.append(entityId)

def test_load_extern_incommon():
    sec_config.xmlsec_binary = sigver.get_xmlsec_binary(["/opt/local/bin"])
    mds = MetadataStore(ONTS.values(), ATTRCONV, sec_config,
                        disable_ssl_certificate_validation=True)

    mds.imp(METADATACONF["10"])
    print(mds)
    assert mds
    assert len(mds.keys())

def test_load_external():
    sec_config.xmlsec_binary = sigver.get_xmlsec_binary(["/opt/local/bin"])
    mds = MetadataStore(ATTRCONV, sec_config,
                        disable_ssl_certificate_validation=True)

    mds.imp(METADATACONF["10"])
    print(mds)
    assert len(mds) == 1  # One source
    assert len(mds.keys()) > 1  # number of idps

def test_ext_2():
    mds = MetadataStore(list(ONTS.values()), ATTRCONV, sec_config, disable_ssl_certificate_validation=True)

    mds.imp(METADATACONF["3"])
    # No specific binding defined

    ents = mds.with_descriptor("spsso")
    for binding in [BINDING_SOAP, BINDING_HTTP_POST, BINDING_HTTP_ARTIFACT, BINDING_HTTP_REDIRECT]:
        assert mds.single_logout_service(list(ents.keys())[0], binding, "spsso")

def test_swamid_idp():
    mds = MetadataStore(ATTRCONV, sec_config,
                        disable_ssl_certificate_validation=True,
                        filter=AllowDescriptor(["idpsso"]))

    mds.imp(METADATACONF["1"])
    sps = mds.with_descriptor("spsso")
    assert len(sps) == 0
    idps = mds.with_descriptor("idpsso")
    assert len(idps) == 275

def test_incommon_1():
    mds = MetadataStore(list(ONTS.values()), ATTRCONV, sec_config, disable_ssl_certificate_validation=True)

    mds.imp(METADATACONF["2"])

    print((mds.entities()))
    assert mds.entities() > 1700
    idps = mds.with_descriptor("idpsso")
    print((list(idps.keys())))
    assert len(idps) > 300  # ~ 18%
    try:
        _ = mds.single_sign_on_service("urn:mace:incommon:uiuc.edu")
    except UnknownPrincipal:
        pass

    idpsso = mds.single_sign_on_service("urn:mace:incommon:alaska.edu")
    assert len(idpsso) == 1
    print(idpsso)
    assert destinations(idpsso) == ["https://idp.alaska.edu/idp/profile/SAML2/Redirect/SSO"]

    sps = mds.with_descriptor("spsso")

    acs_sp = []
    for nam, desc in list(sps.items()):
        if "attribute_consuming_service" in desc:
            acs_sp.append(nam)

    assert len(acs_sp) == 0

    # Look for attribute authorities
    aas = mds.with_descriptor("attribute_authority")

    print((list(aas.keys())))
    assert len(aas) == 180

def test_swami_1():
    UMU_IDP = 'https://idp.umu.se/saml2/idp/metadata.php'
    mds = MetadataStore(ONTS.values(), ATTRCONV, sec_config,
                        disable_ssl_certificate_validation=True)

    mds.imp(METADATACONF["1"])
    assert len(mds) == 1  # One source
    idps = mds.with_descriptor("idpsso")
    assert idps.keys()
    idpsso = mds.single_sign_on_service(UMU_IDP)
    assert len(idpsso) == 1
    assert destinations(idpsso) == [
        'https://idp.umu.se/saml2/idp/SSOService.php']

    _name = name(mds[UMU_IDP])
    assert _name == u'Umeå University (SAML2)'
    certs = mds.certs(UMU_IDP, "idpsso", "signing")
    assert len(certs) == 1

    sps = mds.with_descriptor("spsso")
    assert len(sps) == 108

    wants = mds.attribute_requirement('https://connect8.sunet.se/shibboleth')
    lnamn = [d_to_local_name(mds.attrc, attr) for attr in wants["optional"]]
    assert _eq(lnamn, ['eduPersonPrincipalName', 'mail', 'givenName', 'sn',
                       'eduPersonScopedAffiliation'])

    wants = mds.attribute_requirement('https://beta.lobber.se/shibboleth')
    assert wants["required"] == []
    lnamn = [d_to_local_name(mds.attrc, attr) for attr in wants["optional"]]
    assert _eq(lnamn, ['eduPersonPrincipalName', 'mail', 'givenName', 'sn',
                       'eduPersonScopedAffiliation', 'eduPersonEntitlement'])

def test_metadata():
    conf = config.Config()
    conf.load_file("idp_conf_mdb")
    umu_idp = 'https://idp.umu.se/saml2/idp/metadata.php'
    # Set up a Metadata store
    mds = MetadataStore(ATTRCONV, conf,
                        disable_ssl_certificate_validation=True)

    # Import metadata from local file.
    mds.imp([{"class": "saml2.mdstore.MetaDataFile",
              "metadata": [(full_path("swamid-2.0.xml"), )]}])
    assert len(mds) == 1  # One source

    try:
        export_mdstore_to_mongo_db(mds, "metadata", "test")
    except ConnectionFailure:
        pass
    else:
        mdmdb = MetadataMDB(ATTRCONV, "metadata", "test")
        # replace all metadata instances with this one
        mds.metadata = {"mongo_db": mdmdb}

        idps = mds.with_descriptor("idpsso")
        assert idps.keys()
        idpsso = mds.single_sign_on_service(umu_idp)
        assert len(idpsso) == 1
        assert destinations(idpsso) == [
            'https://idp.umu.se/saml2/idp/SSOService.php']

        _name = name(mds[umu_idp])
        assert _name == u'Ume\xe5 University'
        certs = mds.certs(umu_idp, "idpsso", "signing")
        assert len(certs) == 1

        sps = mds.with_descriptor("spsso")
        assert len(sps) == 417

        wants = mds.attribute_requirement('https://connect.sunet.se/shibboleth')
        assert wants["optional"] == []
        lnamn = [d_to_local_name(mds.attrc, attr) for attr in wants["required"]]
        assert _eq(lnamn,
                   ['eduPersonPrincipalName', 'mail', 'givenName', 'sn',
                    'eduPersonScopedAffiliation', 'eduPersonAffiliation'])

        wants = mds.attribute_requirement(
            "https://gidp.geant.net/sp/module.php/saml/sp/metadata.php/default-sp")
        # Optional
        lnamn = [d_to_local_name(mds.attrc, attr) for attr in wants["optional"]]
        assert _eq(lnamn, ['displayName', 'commonName', 'schacHomeOrganization',
                           'eduPersonAffiliation', 'schacHomeOrganizationType'])
        # Required
        lnamn = [d_to_local_name(mds.attrc, attr) for attr in wants["required"]]
        assert _eq(lnamn, ['eduPersonTargetedID', 'mail',
                           'eduPersonScopedAffiliation'])

 def handleMetadataVerifyJson(self, environ, start_response, qs):
     """
     Handles JSON metadata verifications.
     The post body must contains a JSON message like { 'xml' : 'a metadata file'}
     :param environ: wsgi enviroment
     :param start_response: wsgi start respons
     :param qs: Query parameters in a dictionary.
     :return: wsgi response contaning a JSON response. The JSON message will contain the parameter ok and services.
             ok will contain true if the metadata file can be parsed, otherwise false.
             services will contain a list of all the service names contained in the metadata file.
     """
     ok = False
     services = "[]"
     try:
         if MetadataGeneration.CONST_BODY in qs:
             jsonMessage = json.loads(qs[MetadataGeneration.CONST_BODY])
             if "xml" in jsonMessage:
                 xml = jsonMessage["xml"]
                 xml = xml.strip()
                 metadataOK = False
                 ci = None
                 try:
                     mds = MetadataStore(MetadataGeneration.CONST_ONTS.values(),
                                         MetadataGeneration.CONST_ATTRCONV, self.xmlsec_path,
                                         disable_ssl_certificate_validation=True)
                     md = MetaData(MetadataGeneration.CONST_ONTS.values(), MetadataGeneration.CONST_ATTRCONV, metadata=xml)
                     md.load()
                     entityId = md.entity.keys()[0]
                     mds.metadata[entityId] = md
                     args = {"metad": mds, "dkeys": {"rsa": [self.privateKey]}}
                     ci = utils.ConsumerInfo(['metadata'], **args)
                     metadataOK = True
                 except:
                     self.logger.info('Could not parse the metadata file in handleMetadataVerifyJSON.',
                                       exc_info=True)
                 services = "["
                 first = True
                 if ci is not None:
                     for item in ci._info:
                         if item._ava is not None and entityId in item._ava:
                             for social in item._ava[entityId]:
                                 if not first:
                                     services += ","
                                 else:
                                     first = False
                                 services += '"' + social + '"'
                 services += "]"
                 if metadataOK:
                     ok = True
     except:
         self.logger.fatal('Unknown error in handleMetadataVerifyJSON.',
                           exc_info=True)
     resp = Response('{"ok":"' + str(ok) + '", "services":' + services + '}', headers=[('Content-Type', MetadataGeneration.CONST_TYPEJSON)])
     return resp(environ, start_response)

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 test_switch_1():
    mds = MetadataStore(ONTS.values(), ATTRCONV, sec_config,
                        disable_ssl_certificate_validation=True)

    mds.imp(METADATACONF["5"])
    assert len(mds.keys()) > 160
    idps = mds.with_descriptor("idpsso")
    print(idps.keys())
    idpsso = mds.single_sign_on_service(
        'https://aai-demo-idp.switch.ch/idp/shibboleth')
    assert len(idpsso) == 1
    print(idpsso)
    assert destinations(idpsso) == [
        'https://aai-demo-idp.switch.ch/idp/profile/SAML2/Redirect/SSO']
    assert len(idps) > 30
    aas = mds.with_descriptor("attribute_authority")
    print(aas.keys())
    aad = aas['https://aai-demo-idp.switch.ch/idp/shibboleth']
    print(aad.keys())
    assert len(aad["attribute_authority_descriptor"]) == 1
    assert len(aad["idpsso_descriptor"]) == 1

    sps = mds.with_descriptor("spsso")
    dual = [eid for eid, ent in idps.items() if eid in sps]
    print(len(dual))
    assert len(dual) == 0