Produits Entr'ouvert » w.c.s. » Au quotidien

import urllib.parse

from quixote import get_publisher, get_response, get_request, redirect, get_session

from quixote.directory import Directory

from quixote.html import TemplateIO, htmltext

from wcs.qommon import _

from wcs.qommon.misc import get_variadic_url, simplify

import os

import re

import string

try:

    import lasso

except ImportError:

    pass

from wcs.forms.root import RootDirectory as FormsRootDirectory

from wcs.qommon import N_, get_cfg, get_logger

from wcs.qommon import template

from wcs.qommon import errors

from wcs.qommon.form import *

from wcs.qommon import logger

from wcs.roles import logged_users_role

from wcs.qommon import emails

from wcs.qommon.sms import SMS

from wcs.categories import Category

from wcs.formdef import FormDef

from wcs.data_sources import NamedDataSource

from wcs.qommon.tokens import Token

from wcs.qommon.admin.emails import EmailsDirectory

from wcs.qommon.admin.texts import TextsDirectory

from wcs.workflows import Workflow

import wcs.qommon.ident.password

import wcs.qommon.ident.idp

def category_get_homepage_position(self):

    if hasattr(self, 'homepage_position') and self.homepage_position:

        return self.homepage_position

    if self.url_name == 'consultations':

        return '2nd'

    return '1st'

Category.get_homepage_position = category_get_homepage_position

def category_get_limit(self):

    if hasattr(self, 'limit') and self.limit is not None:

        return self.limit

    return 7

Category.get_limit = category_get_limit

Category.TEXT_ATTRIBUTES = ['name', 'url_name', 'description', 'homepage_position']

Category.INT_ATTRIBUTES = ['position', 'limit']

OldRegisterDirectory = wcs.root.RegisterDirectory

class AlternateRegisterDirectory(OldRegisterDirectory):

    def _q_traverse(self, path):

        return OldRegisterDirectory._q_traverse(self, path)

    def _q_index(self):

        get_logger().info('register')

        ident_methods = get_cfg('identification', {}).get('methods', [])

        if len(ident_methods) == 0:

            idps = get_cfg('idp', {})

            if len(idps) == 0:

                return template.error_page(_('Authentication subsystem is not yet configured.'))

            ident_methods = ['idp']  # fallback to old behaviour; saml.

        if len(ident_methods) == 1:

            method = ident_methods[0]

        else:

            method = 'password'

        return wcs.qommon.ident.register(method)

OldLoginDirectory = wcs.root.LoginDirectory

class AlternateLoginDirectory(OldLoginDirectory):

    def _q_traverse(self, path):

        return OldLoginDirectory._q_traverse(self, path)

    def _q_index(self):

        get_logger().info('login')

        ident_methods = get_cfg('identification', {}).get('methods', [])

        if get_request().form.get('ReturnUrl'):

            get_request().form['next'] = get_request().form.pop('ReturnUrl')

        if 'IsPassive' in get_request().form and 'idp' in ident_methods:

            # if isPassive is given in query parameters, we restrict ourselves

            # to saml login.

            ident_methods = ['idp']

        if len(ident_methods) > 1 and 'idp' in ident_methods:

            # if there is more than one identification method, and there is a

            # possibility of SSO, if we got there as a consequence of an access

            # unauthorized url on admin/ or backoffice/, then idp auth method

            # is chosen forcefully.

            after_url = get_request().form.get('next')

            if after_url:

                root_url = get_publisher().get_root_url()

                after_path = urllib.parse.urlparse(after_url)[2]

                after_path = after_path[len(root_url) :]

                if after_path.startswith(str('admin')) or after_path.startswith(str('backoffice')):

                    ident_methods = ['idp']

        # don't display authentication system choice

        if len(ident_methods) == 1:

            method = ident_methods[0]

            try:

                return wcs.qommon.ident.login(method)

            except KeyError:

                get_logger().error('failed to login with method %s' % method)

                return errors.TraversalError()

        if sorted(ident_methods) == ['idp', 'password']:

            r = TemplateIO(html=True)

            get_response().breadcrumb.append(('login', _('Login')))

            identities_cfg = get_cfg('identities', {})

            form = Form(enctype='multipart/form-data', id='login-form', use_tokens=False)

            if identities_cfg.get('email-as-username', False):

                form.add(StringWidget, 'username', title=_('Email'), size=25, required=True)

            else:

                form.add(StringWidget, 'username', title=_('Username'), size=25, required=True)

            form.add(PasswordWidget, 'password', title=_('Password'), size=25, required=True)

            form.add_submit('submit', _('Connect'))

            if form.is_submitted() and not form.has_errors():

                tmp = wcs.qommon.ident.password.MethodDirectory().login_submit(form)

                if not form.has_errors():

                    return tmp

            r += htmltext('<div id="login-password">')

            r += get_session().display_message()

            r += form.render()

            base_url = get_publisher().get_root_url()

            r += htmltext('<p><a href="%sident/password/forgotten">%s</a></p>') % (

                base_url,

                _('Forgotten password ?'),

            )

            r += htmltext('</div>')

            # XXX: this part only supports a single IdP

            r += htmltext('<div id="login-sso">')

            r += TextsDirectory.get_html_text('aq-sso-text')

            form = Form(enctype='multipart/form-data', action='%sident/idp/login' % base_url)

            form.add_hidden('method', 'idp')

            for kidp, idp in get_cfg('idp', {}).items():

                p = lasso.Provider(

                    lasso.PROVIDER_ROLE_IDP,

                    misc.get_abs_path(idp['metadata']),

                    misc.get_abs_path(idp.get('publickey')),

                    None,

                )

                form.add_hidden('idp', p.providerId)

                break

            form.add_submit('submit', _('Connect'))

            r += form.render()

            r += htmltext('</div>')

            get_request().environ['REQUEST_METHOD'] = 'GET'

            r += htmltext(

                """<script type="text/javascript">

              document.getElementById('login-form')['username'].focus();

            </script>"""

            )

            return r.getvalue()

        else:

            return OldLoginDirectory._q_index(self)

OldIdentDirectory = wcs.root.IdentDirectory

class AlternateIdentDirectory(OldIdentDirectory):

    def _q_traverse(self, path):

        return OldIdentDirectory._q_traverse(self, path)

class AlternatePreviewDirectory(PreviewDirectory):

    def _q_traverse(self, path):

        return super(AlternatePreviewDirectory, self)._q_traverse(path)

    register = AlternateRegisterDirectory()

    login = AlternateLoginDirectory()

    ident = AlternateIdentDirectory()

    preview = AlternatePreviewDirectory()

    def get_substitution_variables(self):

        return {'links': ''}

    def _q_traverse(self, path):

        self.feed_substitution_parts()

        response = get_response()

        if not hasattr(response, 'filter'):

            response.filter = {}

        response.filter['auquotidien'] = True

        if not path or (path[0] not in ('api', 'backoffice') and not get_request().is_json()):

            # api & backoffice have no use for a side box

            response.filter['gauche'] = lambda x: self.box_side(path)

        get_publisher().substitutions.feed(self)

        response.breadcrumb = [('', _('Home'))]

        if not self.admin:

            self.admin = get_publisher().admin_directory_class()

        if not self.backoffice:

            self.backoffice = get_publisher().backoffice_directory_class()

        return super()._q_traverse(path)

    def json(self):

        return FormsRootDirectory().json()

    def categories(self):

        return FormsRootDirectory().categories()

    def _q_index(self):

        if get_request().is_json():

            return FormsRootDirectory().json()

        root_url = get_publisher().get_root_url()

        if get_request().user and get_request().user.anonymous and get_request().user.lasso_dump:

            return redirect('%smyspace/new' % root_url)

        redirect_url = get_cfg('misc', {}).get('homepage-redirect-url')

        if redirect_url:

            return redirect(

                misc.get_variadic_url(redirect_url, get_publisher().substitutions.get_context_variables())

            )

        template.html_top()

        r = TemplateIO(html=True)

        get_response().filter['is_index'] = True

        r += htmltext('<div id="centre">')

        r += self.box_services(position='1st')

        r += htmltext('</div>')

        r += htmltext('<div id="droite">')

        r += self.myspace_snippet()

        r += self.box_services(position='2nd')

        r += self.consultations()

        r += htmltext('</div>')

        user = get_request().user

        if user and user.can_go_in_backoffice():

            get_response().filter['backoffice'] = True

        return r.getvalue()

    def services(self):

        template.html_top()

        return self.box_services(level=2)

    def box_services(self, level=3, position=None):

        ## Services

        if get_request().user and get_request().user.roles:

            accepted_roles = get_request().user.roles

        else:

            accepted_roles = []

        cats = Category.select(order_by='name')

        cats = [x for x in cats if x.url_name != 'consultations']

        Category.sort_by_position(cats)

        all_formdefs = FormDef.select(

            lambda x: not x.is_disabled() or x.disabled_redirection, order_by='name'

        )

        if position:

            t = self.display_list_of_formdefs(

                [x for x in cats if x.get_homepage_position() == position], all_formdefs, accepted_roles

            )

        else:

            t = self.display_list_of_formdefs(cats, all_formdefs, accepted_roles)

        if not t:

            return

        r = TemplateIO(html=True)

        if position == '2nd':

            r += htmltext('<div id="services-2nd">')

        else:

            r += htmltext('<div id="services">')

        if level == 2:

            r += htmltext('<h2>%s</h2>') % _('Services')

        else:

            r += htmltext('<h3>%s</h3>') % _('Services')

        if 'auquotidien-welcome-in-services' in get_response().filter.get('keywords', []):

            homepage_text = TextsDirectory.get_html_text('aq-home-page')

            if homepage_text:

                r += htmltext('<div id="home-page-intro">')

                r += homepage_text

                r += htmltext('</div>')

        r += htmltext('<ul>')

        r += t

        r += htmltext('</ul>')

        r += htmltext('</div>')

        return r.getvalue()

    def display_list_of_formdefs(self, cats, all_formdefs, accepted_roles):

        r = TemplateIO(html=True)

        for category in cats:

            if category.url_name == 'consultations':

                self.consultations_category = category

                continue

            formdefs = [x for x in all_formdefs if str(x.category_id) == str(category.id)]

            formdefs_advertise = []

            for formdef in formdefs[:]:

                if formdef.is_disabled():  # is a redirection

                    continue

                if not formdef.roles:

                    continue

                if not get_request().user:

                    if formdef.always_advertise:

                        formdefs_advertise.append(formdef)

                    formdefs.remove(formdef)

                    continue

                if logged_users_role().id in formdef.roles:

                    continue

                for q in accepted_roles:

                    if q in formdef.roles:

                        break

                else:

                    if formdef.always_advertise:

                        formdefs_advertise.append(formdef)

                    formdefs.remove(formdef)

            if not formdefs and not formdefs_advertise:

                continue

            keywords = {}

            for formdef in formdefs:

                for keyword in formdef.keywords_list:

                    keywords[keyword] = True

            r += htmltext('<li id="category-%s" data-keywords="%s">') % (

                category.url_name,

                ' '.join(keywords),

            )

            r += htmltext('<strong>')

            r += htmltext('<a href="%s/">') % category.url_name

            r += category.name

            r += htmltext('</a></strong>\n')

            r += category.get_description_html_text()

            r += htmltext('<ul>')

            limit = category.get_limit()

            for formdef in formdefs[:limit]:

                r += htmltext('<li data-keywords="%s">') % ' '.join(formdef.keywords_list)

                classes = []

                if formdef.is_disabled() and formdef.disabled_redirection:

                    classes.append('redirection')

                r += htmltext('<a class="%s" href="%s/%s/">%s</a>') % (

                    ' '.join(classes),

                    category.url_name,

                    formdef.url_name,

                    formdef.name,

                )

                r += htmltext('</li>\n')

            if len(formdefs) < limit:

                for formdef in formdefs_advertise[: limit - len(formdefs)]:

                    r += htmltext('<li class="required-authentication">')

                    r += htmltext('<a href="%s/%s/">%s</a>') % (

                        category.url_name,

                        formdef.url_name,

                        formdef.name,

                    )

                    r += htmltext('<span> (%s)</span>') % _('authentication required')

                    r += htmltext('</li>\n')

            if (len(formdefs) + len(formdefs_advertise)) > limit:

                r += htmltext('<li class="all-forms"><a href="%s/" title="%s">%s</a></li>') % (

                    category.url_name,

                    _('Access to all forms of the "%s" category') % category.name,

                    _('Access to all forms in this category'),

                )

            r += htmltext('</ul>')

            r += htmltext('</li>\n')

        return r.getvalue()

    def consultations(self):

        cats = [x for x in Category.select() if x.url_name == 'consultations']

        if not cats:

            return

        consultations_category = cats[0]

        formdefs = FormDef.select(

            lambda x: (

                str(x.category_id) == str(consultations_category.id)

                and (not x.is_disabled() or x.disabled_redirection)

            ),

            order_by='name',

        )

        if not formdefs:

            return

        ## Consultations

        r = TemplateIO(html=True)

        r += htmltext('<div id="consultations">')

        r += htmltext('<h3>%s</h3>') % _('Consultations')

        r += consultations_category.get_description_html_text()

        r += htmltext('<ul>')

        for formdef in formdefs:

            r += htmltext('<li>')

            r += htmltext('<a href="%s/%s/">%s</a>') % (

                consultations_category.url_name,

                formdef.url_name,

                formdef.name,

            )

            r += htmltext('</li>')

        r += htmltext('</ul>')

        r += htmltext('</div>')

        return r.getvalue()

    def box_side(self, path):

        r = TemplateIO(html=True)

        root_url = get_publisher().get_root_url()

        if (

            path == ['']

            and 'include-tracking-code-form' in get_response().filter.get('keywords', [])

            and self.has_anonymous_access_codes()

        ):

            r += htmltext('<form id="follow-form" action="%scode/load">') % root_url

            r += htmltext('<h3>%s</h3>') % _('Tracking code')

            r += htmltext('<input size="12" name="code" placeholder="%s"/>') % _('ex: RPQDFVCD')

            r += htmltext('<input type="submit" value="%s"/>') % _('Load')

            r += htmltext('</form>')

        cats = Category.select(order_by='name')

        cats = [x for x in cats if x.url_name != 'consultations' and x.get_homepage_position() == 'side']

        Category.sort_by_position(cats)

        if cats:

            r += htmltext('<div id="side-services">')

            r += htmltext('<h3>%s</h3>') % _('Services')

            r += htmltext('<ul>')

            for cat in cats:

                r += htmltext('<li><a href="%s/">%s</a></li>') % (cat.url_name, cat.name)

            r += htmltext('</ul>')

            r += htmltext('</div>')

        v = r.getvalue()

        if v:

            r = TemplateIO(html=True)

            r += htmltext('<div id="sidebox">')

            r += v

            r += htmltext('</div>')

            return r.getvalue()

        return None

    def has_anonymous_access_codes(self):

        return any((x for x in FormDef.select() if x.enable_tracking_codes))

    def myspace_snippet(self):

        r = TemplateIO(html=True)

        r += htmltext('<div id="myspace">')

        r += htmltext('<h3>%s</h3>') % _('My Space')

        r += htmltext('<ul>')

        if get_request().user and not get_request().user.anonymous:

            r += htmltext('  <li><a href="myspace/" id="member">%s</a></li>') % _(

                'Access to your personal space'

            )

            r += htmltext('  <li><a href="logout" id="logout">%s</a></li>') % _('Logout')

        else:

            r += htmltext('  <li><a href="register/" id="inscr">%s</a></li>') % _('Registration')

            r += htmltext('  <li><a href="login/" id="login">%s</a></li>') % _('Login')

        r += htmltext('</ul>')

        r += htmltext('</div>')

        return r.getvalue()

get_publisher_class().root_directory_class = AlternateRootDirectory

TextsDirectory.register(

    'aq-sso-text',

    N_('Connecting with Identity Provider'),

    default=N_(

        '''<h3>Connecting with Identity Provider</h3>

<p>You can also use your identity provider to connect.

</p>'''

    ),

)

TextsDirectory.register('aq-home-page', N_('Home Page'), wysiwyg=True)