Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.openid.openid_manager

"""
Manage the OpenID consumer and related data stores.
"""

import logging
import os
import pickle

try:
    from openid import oidutil
    from openid.consumer import consumer
    from openid.extensions import sreg
    from openid.store import filestore
except ImportError:
    oidutil = None

    class FakeConsumer(object):
        def __getattr__(x, y):
            return None
    consumer = FakeConsumer()


OPENID_IMPORT_MESSAGE = ('The Python openid package is required to use this '
                         'feature, please install it')

log = logging.getLogger(__name__)


[docs]def oidlog(message, level=0): log.debug(message)
if oidutil is not None: oidutil.log = oidlog
[docs]class OpenIDManager(object):
[docs] def __init__(self, cache_path): assert oidutil is not None, OPENID_IMPORT_MESSAGE self.session_path = os.path.join(cache_path, 'session') self.store_path = os.path.join(cache_path, 'store') for dir in self.session_path, self.store_path: if not os.path.exists(dir): os.makedirs(dir) self.store = filestore.FileOpenIDStore(self.store_path)
[docs] def get_session(self, trans): session_file = os.path.join(self.session_path, str(trans.galaxy_session.id)) if not os.path.exists(session_file): pickle.dump(dict(), open(session_file, 'wb')) return pickle.load(open(session_file, 'rb'))
[docs] def persist_session(self, trans, oidconsumer): session_file = os.path.join(self.session_path, str(trans.galaxy_session.id)) pickle.dump(oidconsumer.session, open(session_file, 'wb'))
[docs] def get_consumer(self, trans): return consumer.Consumer(self.get_session(trans), self.store)
[docs] def add_sreg(self, trans, request, required=None, optional=None): if required is None: required = [] if optional is None: optional = [] sreg_request = sreg.SRegRequest(required=required, optional=optional) request.addExtension(sreg_request)
[docs] def get_sreg(self, info): return sreg.SRegResponse.fromSuccessResponse(info)
# so I don't have to expose all of openid.consumer.consumer FAILURE = consumer.FAILURE SUCCESS = consumer.SUCCESS CANCEL = consumer.CANCEL SETUP_NEEDED = consumer.SETUP_NEEDED