Produits Entr'ouvert » django-mellon

A list of dictionaries, they must contain at least one of the keys `METADATA`

(inline copy of the identity provider metadata), `METADATA_URL` URL of the IdP

metadata file, or `METADATA_PATH` an absolute path to the IdP metadata file..

All other keys are override of generic settings.

When using an URL, the URL is automatically cached in the `MEDIA_ROOT`

directory of your application in the directory named `mellon_metadata_cache`.

If you restart the application and the URL is unavailable, the file cache will

be used. The cache will be refreshed every `MELLON_METADATA_CACHE_TIME` seconds.

If the HTTP retrieval of the metadata URL takes longer thant

`METTON_METADATA_HTTP_TIMEOUT` seconds, retrieval will be skipped.

When the cache is already loaded, retrievals are done in the background by a

thread.

When using a local absolute path, the metadata is reloaded each time the

modification time of the file is superior to the last time it was loaded.

MELLON_METADATA_CACHE_TIME

--------------------------

When using METADATA_URL to reference a metadata file, it's the duration in

secondes between refresh of the metadata file. Default is 3600 seconds, 1 hour.

METTON_METADATA_HTTP_TIMEOUT

---------------------------

Timeout in seconds for HTTP call made to retrieve metadata files. Default is 10

seconds.

    python-lasso,

    python-atomicwrites

    python3-lasso,

    python3-atomicwrites

from xml.etree import ElementTree as ET

import hashlib

import os

import threading

import time

from atomicwrites import atomic_write

from django.core.files.storage import default_storage

from django.utils.six.moves.urllib.parse import urlparse

logger = logging.getLogger(__name__)

            if self.load_idp(idp, i):

                yield idp

    def load_metadata_path(self, idp, i):

        path = idp['METADATA_PATH']

        if not os.path.exists(path):

            logger.warning('metadata path %s does not exist', path)

            return

        last_update = idp.get('METADATA_PATH_LAST_UPDATE', 0)

        try:

            mtime = os.stat(path).st_mtime

        except OSError as e:

            logger.warning('metadata path %s : stat() call failed, %s', path, e)

            return

        if last_update == 0 or mtime >= last_update:

            idp['METADATA_PATH_LAST_UPDATE'] = time.time()

            try:

                with open(path) as fd:

                    metadata = fd.read()

            except OSError as e:

                logger.warning('metadata path %s : open()/read() call failed, %s', path, e)

                return

            entity_id = self.load_entity_id(metadata, i)

            if not entity_id:

                logger.error('invalid metadata file retrieved from %s', path)

                return

            if 'ENTITY_ID' in idp and idp['ENTITY_ID'] != entity_id:

                logger.error('metadata path %s : entityID changed %r != %r', path, entity_id, idp['ENTITY_ID'])

                del idp['ENTITY_ID']

            idp['METADATA'] = metadata

    def load_metadata_url(self, idp, i):

        url = idp['METADATA_URL']

        try:

            hostname = urlparse(url).hostname

        except (ValueError, TypeError) as e:

            logger.error('invalid METADATA_URL %r: %s', url, e)

            return

        if not hostname:

            logger.error('no hostname in METADATA_URL %r: %s', url)

            return

        last_update = idp.get('METADATA_URL_LAST_UPDATE', 0)

        metadata_cache_time = utils.get_setting(idp, 'METADATA_CACHE_TIME')

        timeout = utils.get_setting(idp, 'METADATA_HTTP_TIMEOUT')

        now = time.time()

        try:

            url_fingerprint = hashlib.md5(url.encode('ascii')).hexdigest()

            file_cache_key = '%s_%s.xml' % (hostname, url_fingerprint)

        except (UnicodeError, TypeError, ValueError):

            logger.exception('unable to compute file_cache_key')

            return

        cache_directory = default_storage.path('mellon_metadata_cache')

        file_cache_path = os.path.join(cache_directory, file_cache_key)

        if not os.path.exists(cache_directory):

            os.makedirs(cache_directory)

        if os.path.exists(file_cache_path) and 'METADATA' not in idp:

            try:

                with open(file_cache_path) as fd:

                    idp['METADATA'] = fd.read()

            except OSError:

                pass

        # fresh cache, skip loading

        if last_update and 'METADATA' in idp and (now - last_update) < metadata_cache_time:

            return

        def __http_get():

            try:

                    response = requests.get(url, verify=verify_ssl_certificate, timeout=timeout)

                        url, e, i)

                    try:

                        with open(file_cache_path) as fd:

                            pass

                        idp['METADATA_PATH'] = file_cache_path

                        self.load_metadata_path()

                    except IOError:

                        pass

                    return

                entity_id = self.load_entity_id(response.text, i)

                if not entity_id:

                    logger.error('invalid metadata file retrieved from %s', url)

                    return

                if 'ENTITY_ID' in idp and idp['ENTITY_ID'] != entity_id:

                    logger.error('metadata url %s : entityID changed %r != %r', url, entity_id, idp['ENTITY_ID'])

                    del idp['ENTITY_ID']

                idp['METADATA_URL_LAST_UPDATE'] = now

                with atomic_write(file_cache_path, mode='wb', overwrite=True) as fd:

                    fd.write(response.text.encode('utf-8'))

                idp['METADATA_PATH'] = file_cache_path

                idp['METADATA_PATH_LAST_UPDATE'] = time.time() + 1

                idp.pop('METADATA_URL_UPDATE_THREAD', None)

                logger.debug('metadata url %s : update throught HTTP', url)

            finally:

                stale_timeout = 24 * metadata_cache_time

                if last_update and (now - idp['METADATA_URL_LAST_UPDATE']) > stale_timeout:

                    logger.error('metadata url %s : not updated since %.1f hours',

                                 stale_timeout / 3600.0)

        # we have cache, update in background

        if last_update and 'METADATA' in idp:

            t = threading.Thread(target=__http_get)

            t.start()

            # suspend updates for HTTP timeout + 5 seconds

            idp['METADATA_URL_UPDATE_THREAD'] = t

            idp['METADATA_URL_LAST_UPDATE'] = last_update + timeout + 5

        else:

            # synchronous update

            __http_get()

    def load_metadata(self, idp, i):

        # legacy support

        if 'METADATA' in idp and idp['METADATA'].startswith('/'):

            idp['METADATA_PATH'] = idp['METADATA']

            del idp['METADATA']

        if 'METADATA_PATH' in idp:

            self.load_metadata_path(idp, i)

        if 'METADATA_URL' in idp:

            self.load_metadata_url(idp, i)

        if 'METADATA' in idp:

                entity_id = self.load_entity_id(idp['METADATA'], i)

                if entity_id:

                    idp['ENTITY_ID'] = entity_id

            return idp['METADATA']

    def load_entity_id(self, metadata, i):

        try:

            doc = ET.fromstring(metadata)

        except (TypeError, ET.ParseError):

            self.logger.error(u'METADATA of %d-th idp is invalid', i)

            return None

        if doc.tag != '{%s}EntityDescriptor' % lasso.SAML2_METADATA_HREF:

            self.logger.error(u'METADATA of %d-th idp has no EntityDescriptor root tag', i)

            return None

        if 'entityID' not in doc.attrib:

            self.logger.error(

                u'METADATA of %d-th idp has no entityID attribute on its root tag', i)

            return None

        return doc.attrib['entityID']

    def load_idp(self, idp, i):

        metadata = self.load_metadata(idp, i)

        if not metadata:

            self.logger.error(u'missing METADATA or METADATA_URL in %d-th idp', i)

            return False

        return 'ENTITY_ID' in idp

        'METADATA_CACHE_TIME': 3600,

        'METADATA_HTTP_TIMEOUT': 10,

    try:

        parser.Parse(content, True)

    except expat.ExpatError as e:

        raise ValueError('invalid XML %s' % e)

          'atomicwrites',

import os

@pytest.fixture(autouse=True)

def settings(settings, tmpdir):

    settings.MEDIA_ROOT = str(tmpdir.mkdir('media'))

    return settings

def app(request, settings):

def private_settings(request, tmpdir):

@pytest.fixture(scope='session')

def metadata():

    with open(os.path.join(os.path.dirname(__file__), 'metadata.xml')) as fd:

        yield fd.read()

@pytest.fixture

def metadata_path(tmpdir, metadata):

    metadata_path = tmpdir / 'metadata.xml'

    with metadata_path.open('w') as fd:

        fd.write(metadata)

    yield str(metadata_path)

import datetime

import time

import pytest

@pytest.fixture

def adapter():

    return DefaultAdapter()

def test_load_metadata_simple(adapter, metadata):

    idp = {'METADATA': metadata}

    assert adapter.load_metadata(idp, 0) == metadata

def test_load_metadata_legacy(adapter, metadata_path, metadata):

    idp = {'METADATA': metadata_path}

    assert adapter.load_metadata(idp, 0) == metadata

    assert idp['METADATA'] == metadata

def test_load_metadata_path(adapter, metadata_path, metadata, freezer):

    now = time.time()

    idp = {'METADATA_PATH': str(metadata_path)}

    assert adapter.load_metadata(idp, 0) == metadata

    assert idp['METADATA'] == metadata

    assert idp['METADATA_PATH_LAST_UPDATE'] == now

def test_load_metadata_url(settings, adapter, metadata, httpserver, freezer, caplog):

    now = time.time()

    httpserver.serve_content(content=metadata, headers={'Content-Type': 'application/xml'})

    idp = {'METADATA_URL': httpserver.url}

    assert adapter.load_metadata(idp, 0) == metadata

    assert idp['METADATA'] == metadata

    assert idp['METADATA_URL_LAST_UPDATE'] == now

    assert 'METADATA_PATH' in idp

    assert idp['METADATA_PATH'].startswith(settings.MEDIA_ROOT)

    with open(idp['METADATA_PATH']) as fd:

        assert fd.read() == metadata

    assert idp['METADATA_PATH_LAST_UPDATE'] == now + 1

    httpserver.serve_content(content=metadata.replace('idp5', 'idp6'),

                             headers={'Content-Type': 'application/xml'})

    assert adapter.load_metadata(idp, 0) == metadata

    freezer.move_to(datetime.timedelta(seconds=3601))

    caplog.clear()

    assert adapter.load_metadata(idp, 0) == metadata

    # wait for update thread to finish

    try:

        idp['METADATA_URL_UPDATE_THREAD'].join()

    except KeyError:

        pass

    new_meta = adapter.load_metadata(idp, 0)

    assert new_meta != metadata

    assert new_meta == metadata.replace('idp5', 'idp6')

    assert 'entityID changed' in caplog.records[-1].message

    assert caplog.records[-1].levelname == 'ERROR'

    # test load from file cache

    del idp['METADATA']

    del idp['METADATA_PATH']

    del idp['METADATA_PATH_LAST_UPDATE']

    httpserver.serve_content(content='', headers={'Content-Type': 'application/xml'})

    assert adapter.load_metadata(idp, 0) == metadata.replace('idp5', 'idp6')

def test_create_server_good_metadata_file(mocker, rf, private_settings, tmpdir, caplog):

    path = tmpdir / 'metadata.xml'

    with path.open('w') as fd:

        fd.write(open('tests/metadata.xml').read())

            'METADATA': str(path),

    server = create_server(request)

envlist = coverage-py2-{dj18,dj111}-{pg,sqlite},coverage-py3-dj111-{pg,sqlite}

  pytest-freezegun

  pytest-localserver