Produits Entr'ouvert » Larpe

import libxml2

import urllib

import urlparse

import httplib

import re

import os

from quixote import get_field, get_request, get_response, get_session, get_session_manager, redirect

from quixote.directory import Directory

from quixote.http_request import parse_header

import lasso

from qommon import get_logger

from qommon.form import *

from qommon.template import *

from qommon.liberty import soap_call, SOAPException

import misc

from users import User

from hosts import Host

from federations import Federation

import site_authentication

class Liberty(Directory):

    _q_exports = ['', 'login', 'assertionConsumer', 'soapEndpoint',

            'singleLogout', 'singleLogoutReturn',

            'federationTermination', 'federationTerminationReturn',

            ('metadata.xml', 'metadata'), 'public_key', 'local_auth']

    def perform_login(self, idp = None):

        server = misc.get_lasso_server()

        login = lasso.Login(server)

        login.initAuthnRequest(idp, lasso.HTTP_METHOD_REDIRECT)

        login.request.nameIdPolicy = 'federated'

        login.request.forceAuthn = False

        login.request.isPassive = False

        login.request.consent = 'urn:liberty:consent:obtained'

        login.buildAuthnRequestMsg()

        return redirect(login.msgUrl)

    def assertionConsumer(self):

        server = misc.get_lasso_server()

        if not server:

            return error_page(_('Liberty support is not yet configured'))

        login = lasso.Login(server)

        request = get_request()

        if request.get_method() == 'GET' or get_field('LAREQ'):

            if request.get_method() == 'GET':

                login.initRequest(request.get_query(), lasso.HTTP_METHOD_REDIRECT)

            else:

                login.initRequest(get_field('LAREQ'), lasso.HTTP_METHOD_POST)

            login.buildRequestMsg()

            try:

                soap_answer = soap_call(login.msgUrl, login.msgBody)

            except SOAPException:

                return error_page(_('Failure to communicate with identity provider'))

            try:

                login.processResponseMsg(soap_answer)

            except lasso.Error, error:

                if error[0] == lasso.LOGIN_ERROR_STATUS_NOT_SUCCESS:

                    return error_page(_('Unknown authentication failure'))

                if hasattr(lasso, 'LOGIN_ERROR_UNKNOWN_PRINCIPAL'):

                    if error[0] == lasso.LOGIN_ERROR_UNKNOWN_PRINCIPAL:

                        return error_page(_('Authentication failure; unknown principal'))

                return error_page(_("Identity Provider didn't accept artifact transaction."))

        else:

            login.processAuthnResponseMsg(get_field('LARES'))

        login.acceptSso()

        session = get_session()

        if login.isSessionDirty:

            if login.session:

                session.lasso_session_dumps[server.providerId] = login.session.dump()

            else:

                session.lasso_session_dumps[server.providerId] = None

        # Look for an existing user

        user = self.lookup_user(session, login)

        # Check if it is for Larpe administration or token

        host = Host.get_host_from_url()

        if host is None:

            return redirect('%s/' % get_request().environ['SCRIPT_NAME'])

        if host.name == 'larpe':

            if user:

                session.set_user(user.id, server.providerId)

            else:

                session.name_identifier = login.nameIdentifier.content

                session.lasso_anonymous_identity_dump = login.identity.dump()

                session.provider_id = server.providerId

            if session.after_url:

                # Access to an admin page or token url with parameter

                after_url = session.after_url

                session.after_url = None

                return redirect(after_url)

            if user and user.is_admin:

                return redirect('%s/admin/' % get_request().environ['SCRIPT_NAME'])

            else:

                return redirect('%s/token' % get_request().environ['SCRIPT_NAME'])

        # Set session user

        if not user:

            user = User()

        user.name_identifiers = [ login.nameIdentifier.content ]

        user.lasso_dumps = [ login.identity.dump() ]

        user.store()

        session.set_user(user.id, server.providerId)

        federations = Federation.select(lambda x: host.id == x.host_id \

                                        and user.name_identifiers[0] in x.name_identifiers)

        if federations:

            return site_authentication.get_site_authentication(host).sso_local_login(federations[0])

        else:

            response = get_response()

            if session.after_url:

                after_url = session.after_url

                session.after_url = None

                return redirect(after_url)

            response.set_status(303)

            response.headers['location'] = urlparse.urljoin(request.get_url(), str('local_auth'))

            response.content_type = 'text/plain'

            return 'Your browser should redirect you'

    def lookup_user(self, session, login):

        found_users = list(User.select(lambda x: login.nameIdentifier.content in x.name_identifiers))

        if found_users:

            return found_users[0]

        return None

    def singleLogout(self):

        request = get_request()

        logout = lasso.Logout(misc.get_lasso_server())

        if lasso.isLibertyQuery(request.get_query()):

            try:

                logout.processRequestMsg(request.get_query())

            except lasso.Error, error:

                if error[0] == lasso.DS_ERROR_INVALID_SIGNATURE:

                    return error_page(_('Failed to check single logout request signature.'))

                raise

            session = get_session()

            if not session.id:

  	            # session has not been found, this may be because the user has

  	            # its browser configured so that cookies are not sent for

  	            # remote queries and IdP is using image-based SLO.

  	            # so we look up a session with the appropriate name identifier

                for session in get_session_manager().values():

                    # This block differs from qommon

                    user = session.get_user(logout.server.providerId)

                    if user and logout.nameIdentifier.content in user.name_identifiers:

                        break

                else:

                    session = get_session()

            return self.slo_idp(logout, session)

        else:

            return self.slo_sp(logout, get_session())

    def singleLogoutReturn(self):

        logout = lasso.Logout(misc.get_lasso_server())

        host = Host.get_host_from_url()

        if host is None:

            return redirect('%s/' % get_request().environ['SCRIPT_NAME'])

        try:

            logout.processResponseMsg(get_request().get_query())

        except lasso.Error, error:

            if error[0] == lasso.PROFILE_ERROR_INVALID_QUERY:

                raise AccessError()

            if error[0] == lasso.DS_ERROR_INVALID_SIGNATURE:

                return error_page(_('Failed to check single logout request signature.'))

            if hasattr(lasso, 'LOGOUT_ERROR_REQUEST_DENIED') and \

                    error[0] == lasso.LOGOUT_ERROR_REQUEST_DENIED:

                return redirect(host.get_root_url()) # ignore silently

            elif error[0] == lasso.ERROR_UNDEFINED:

                # XXX: unknown status; ignoring for now.

                return redirect(host.get_root_url()) # ignore silently

            raise

        return redirect(host.get_root_url())

    def slo_idp(self, logout, session):

        '''Single Logout initiated by IdP'''

        # This block differs from qommon

        if session.lasso_session_dumps.has_key(logout.server.providerId):

            logout.setSessionFromDump(session.lasso_session_dumps[logout.server.providerId])

        user = session.get_user(logout.server.providerId)

        if user and user.lasso_dumps:

            logout.setIdentityFromDump(user.lasso_dumps[0])

        if user and logout.nameIdentifier.content not in user.name_identifiers:

            raise 'No appropriate name identifier in user (%s and %s)' % (

                    logout.nameIdentifier.content, user.name_identifiers)

        host = Host.get_host_with_provider_id(logout.server.providerId)

        if host is not None:

            site_authentication.get_site_authentication(host).local_logout(user=user)

        try:

            logout.validateRequest()

        except lasso.Error, error:

            if error[0] != lasso.PROFILE_ERROR_SESSION_NOT_FOUND:

                raise

        else:

            get_session_manager().expire_session(logout.server.providerId)

        logout.buildResponseMsg()

        if logout.msgBody: # soap answer

            return logout.msgBody

        else:

            return redirect(logout.msgUrl)

    def slo_sp(self, logout, session):

        host = Host.get_host_from_url()

        if host is None:

            return redirect('%s/' % get_request().environ['SCRIPT_NAME'])

        if not session.id or not session.users.has_key(logout.server.providerId) \

                or not session.lasso_session_dumps.has_key(logout.server.providerId):

            get_session_manager().expire_session(logout.server.providerId)

            return redirect(host.get_root_url())

        logout.setSessionFromDump(session.lasso_session_dumps[logout.server.providerId])

        user = session.get_user(logout.server.providerId)

        if host.name != 'larpe' and user:

            site_authentication.get_site_authentication(host).local_logout(user=user)

        if user and user.lasso_dumps:

            logout.setIdentityFromDump(user.lasso_dumps[0])

        else:

            get_session_manager().expire_session(logout.server.providerId)

            return redirect(host.get_root_url())

        return self.slo_sp_redirect(logout, host)

    def slo_sp_redirect(self, logout, host):

        try:

            logout.initRequest(None, lasso.HTTP_METHOD_REDIRECT)

        except lasso.Error, error:

            if error[0] == lasso.PROFILE_ERROR_NAME_IDENTIFIER_NOT_FOUND:

                get_session_manager().expire_session()

                return redirect(host.get_root_url())

            raise

        logout.buildRequestMsg()

        get_session_manager().expire_session(logout.server.providerId)

        return redirect(logout.msgUrl)

    def soapEndpoint(self):

        request = get_request()

        ctype = request.environ.get('CONTENT_TYPE')

        if not ctype:

            return

        ctype, ctype_params = parse_header(ctype)

        if ctype != 'text/xml':

            return

        response = get_response()

        response.set_content_type('text/xml')

        length = int(request.environ.get('CONTENT_LENGTH'))

        soap_message = request.stdin.read(length)

        request_type = lasso.getRequestTypeFromSoapMsg(soap_message)

        if request_type == lasso.REQUEST_TYPE_LOGOUT:

            logout = lasso.Logout(misc.get_lasso_server())

            logout.processRequestMsg(soap_message)

            name_identifier = logout.nameIdentifier.content

            for session in get_session_manager().values():

                user = session.get_user(logout.server.providerId)

                if user and logout.nameIdentifier.content in user.name_identifiers:

                    break

            else:

                session = None

            return self.slo_idp(logout, session)

        if request_type == lasso.REQUEST_TYPE_DEFEDERATION:

            defederation = lasso.Defederation(misc.get_lasso_server())

            defederation.processNotificationMsg(soap_message)

            for session in get_session_manager().values():

                user = session.get_user(defederation.server.providerId)

                if user and defederation.nameIdentifier.content in user.name_identifiers:

                    break

            else:

                session = None

            return self.fedterm(defederation, session)

    def federationTermination(self):

        request = get_request()

        if not lasso.isLibertyQuery(request.get_query()):

            return redirect('%s/' % get_request().environ['SCRIPT_NAME'])

        defederation = lasso.Defederation(misc.get_lasso_server())

        defederation.processNotificationMsg(request.get_query())

        return self.fedterm(defederation, get_session())

    def fedterm(self, defederation, session):

        if session is not None:

            host = Host.get_host_with_provider_id(defederation.server.providerId)

            if host is not None:

                site_authentication.get_site_authentication(host).local_defederate(session, defederation.server.providerId)

            if session.lasso_session_dumps.has_key(defederation.server.providerId):

                defederation.setSessionFromDump(session.lasso_session_dumps[defederation.server.providerId])

            user = session.get_user(defederation.server.providerId)

            if user and user.lasso_dumps:

                defederation.setIdentityFromDump(user.lasso_dumps[0])

        else:

            user = None

        try:

            defederation.validateNotification()

        except lasso.Error, error:

            pass # ignore failure (?)

        else:

            if user:

                if not defederation.identity:

                    # if it was the last federation the whole identity dump collapsed

                    del user.lasso_dumps[0]

                else:

                    user.lasso_dumps[0] = defederation.identity.dump()

                user.store()

        if user and defederation.nameIdentifier.content:

            user.remove_name_identifier(defederation.nameIdentifier.content)

            user.store()

        if defederation.isSessionDirty and session is not None:

            if not defederation.session:

                del session.lasso_session_dumps[defederation.server.providerId]

            else:

                session.lasso_session_dumps[defederation.server.providerId] = defederation.session.dump()

            session.store()

        get_session_manager().expire_session(defederation.server.providerId)

        if defederation.msgUrl:

            return redirect(defederation.msgUrl)

        else:

            response = get_response()

            response.set_status(204)

            return ''

    def federationTerminationReturn(self):

        host = Host.get_host_from_url()

        if host is None:

            return redirect('%s/' % get_request().environ['SCRIPT_NAME'])

        return redirect(host.get_return_url())

    def local_auth(self):

        host = Host.get_host_from_url()

        if host is None:

            return redirect('%s/' % get_request().environ['SCRIPT_NAME'])

        return site_authentication.get_site_authentication(host).local_auth

    local_auth = property(local_auth)

    def metadata(self):

        host = Host.get_host_from_url()

        if host is None:

            return redirect('%s/' % get_request().environ['SCRIPT_NAME'])

        get_response().set_content_type('text/xml', 'utf-8')

        metadata = unicode(open(host.metadata).read(), 'utf-8')

        return metadata

    def public_key(self):

        host = Host.get_host_from_url()

        if host is None:

            return redirect('%s/' % get_request().environ['SCRIPT_NAME'])

        get_response().set_content_type('text/plain')

        public_key = open(host.public_key).read()

        return public_key