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']

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

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

        warning = logger.warning

        if 'METADATA' not in idp:

            # if we have no metadata in cache, we must emit errors

            warning = logger.error

        try:

            hostname = urlparse(url).hostname

        except (ValueError, TypeError) as e:

            warning('invalid METADATA_URL %r: %s', url, e)

            return

        if not hostname:

            warning('no hostname in METADATA_URL %r: %s', url)

            return

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

        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):

            warning('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 metadata_cache_time:

            # METADATA_CACHE_TIME == 0 disable the file cache

            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()

                    # use file cache mtime as last_update time, prevent too many loading from different workers

                    last_update = max(last_update, os.stat(file_cache_path).st_mtime)

                except OSError:

                    warning('metadata url %s : error when loading the file cache %s', url, file_cache_path)

        # 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)

                    warning('metadata url %s : HTTP request failed %s', url, e)

                    return

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

                if not entity_id:

                    warning('invalid metadata file retrieved from %s', url)

                    return

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

                    # entityID change is always en error

                    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

                if metadata_cache_time:

                    try:

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

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

                    except OSError as e:

                        logger.error('metadata url %s : could not write file cache %s, %s', url, file_cache_path, e)

                idp['METADATA_PATH'] = file_cache_path

                # prevent reloading of the file cache immediately

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

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

            finally:

                # release thread object

                idp.pop('METADATA_URL_UPDATE_THREAD', None)

                # emit an error if cache is too old

                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()

            # store thread in idp for tests

            idp['METADATA_URL_UPDATE_THREAD'] = t

            # suspend updates for HTTP timeout + 5 seconds

            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

        if 'ENTITY_ID' in idp:

            return idp['METADATA']

    def load_entity_id(self, metadata, i):

        try:

            doc = ET.fromstring(metadata)

        except (TypeError, ET.ParseError):

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

            return None

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

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

            return None

        if 'entityID' not in doc.attrib:

            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):

        self.load_metadata(idp, i)

        return 'ENTITY_ID' in idp

            logger.error(u'invalid username template %r', username_template)

            logger.error(

            logger.exception(u'unknown error when formatting username')

            logger.warning('could not build a username, login refused')

                        logger.warning('more than one value for attribute %r, cannot federate',

                                       transient_federation_attribute)

            logger.info('looked up user %s with name_id %s from issuer %s', user, name_id, issuer)

                logger.debug('provisionning disabled, login refused')

            logger.info('looked up user %s with name_id %s from issuer %s', nameid_user, name_id, issuer)

            logger.info('created new user %s with name_id %s from issuer %s', nameid_user, name_id, issuer)

            logger.error('invalid LOOKUP_BY_ATTRIBUTES configuration %r: it must be a list', lookup_by_attributes)

                logger.error('invalid LOOKUP_BY_ATTRIBUTES configuration %r: it must be a list of dicts', line)

                logger.error('invalid LOOKUP_BY_ATTRIBUTES configuration %r: user_field is missing', line)

                logger.error('invalid LOOKUP_BY_ATTRIBUTES configuration %r, user field %s does not exist',

                             line, user_field)

                logger.error('invalid LOOKUP_BY_ATTRIBUTES configuration %r: saml_attribute is missing', line)

                logger.error('looking for user by saml attribute %r and user field %r, skipping because empty',

                             saml_attribute, user_field)

                    logger.debug('looking for users by attribute %r and user field %r with value %r: not found',

                                 saml_attribute, user_field, value)

                logger.info(u'looking for user by attribute %r and user field %r with value %r: found %s',

                            saml_attribute, user_field, value, display_truncated_list(users_found))

            logger.info(u'looking for user by attributes %r: found user %s', lookup_by_attributes, user)

            logger.warning(u'looking for user by attributes %r: too many users found(%d), failing',

                           lookup_by_attributes, len(users))

                logger.warning(u'invalid attribute mapping template %r', tpl)

                logger.warning(

                    logger.info(u'set field %s of user %s to value %r (old value %r)', field, user, value, old_value)

                        logger.info('flag is_staff and is_superuser added to user %s', user)

                logger.info('flag is_staff and is_superuser removed from user %s', user)

                logger.info(

                logger.info(u'removing group %s (%s) from user %s (%s)', rel.group, rel.group.pk, rel.user, rel.user.pk)

        'METADATA_CACHE_TIME': 3600,

        'METADATA_HTTP_TIMEOUT': 10,

            if idp and idp.get('METADATA'):

                try:

                    server.addProviderFromBuffer(lasso.PROVIDER_ROLE_IDP, idp['METADATA'])

                except lasso.Error as e:

                    logger.error(u'bad metadata in idp %s, %s', idp['ENTITY_ID'], e)

        if not idp:

            self.log.warning('entity id %r is unknown', login.remoteProviderId)

            return HttpResponseBadRequest(

                'entity id %r is unknown' % login.remoteProviderId)

        if not idp:

          '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

    assert 'flag is_staff and is_superuser removed' not in caplog.text

@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_load_metadata_url_stale_timeout(settings, adapter, metadata, httpserver, freezer, caplog):

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

    idp = {'METADATA_URL': httpserver.url}

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

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

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

    freezer.move_to(datetime.timedelta(seconds=24 * 3600 - 1))

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

    # wait for update thread to finish

    try:

        idp['METADATA_URL_UPDATE_THREAD'].join()

    except KeyError:

        pass

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

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

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

    # wait for update thread to finish

    try:

        idp['METADATA_URL_UPDATE_THREAD'].join()

    except KeyError:

        pass

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

from mellon.utils import create_metadata, iso8601_to_datetime, flatten_datetime

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

  pytest-freezegun

  pytest-localserver