Produits Entr'ouvert » w.c.s.

from wcs.wf.register_comment import RegisterCommenterWorkflowStatusItem

def test_workflow_global_webservice_trigger(pub, local_user):

    workflow = Workflow(name='test')

    st1 = workflow.add_status('Status1', 'st1')

    ac1 = workflow.add_global_action('Action', 'ac1')

    trigger = ac1.append_trigger('webservice')

    trigger.identifier = 'plop'

    add_to_journal = RegisterCommenterWorkflowStatusItem()

    add_to_journal.id = '_add_to_journal'

    add_to_journal.comment = 'HELLO WORLD'

    ac1.items.append(add_to_journal)

    add_to_journal.parent = ac1

    workflow.store()

    FormDef.wipe()

    formdef = FormDef()

    formdef.name = 'test'

    formdef.fields = []

    formdef.workflow_id = workflow.id

    formdef.store()

    formdef.data_class().wipe()

    formdata = formdef.data_class()()

    formdata.just_created()

    formdata.store()

    assert formdef.data_class().get(formdata.id).status == 'wf-st1'

    # call to undefined hook

    resp = get_app(pub).post(sign_uri(formdata.get_url() + 'hooks/XXX/'), status=404)

    resp = get_app(pub).post(sign_uri(formdata.get_api_url() + 'hooks/XXX/'), status=404)

    # anonymous call

    resp = get_app(pub).post(formdata.get_url() + 'hooks/plop/', status=200)

    assert formdef.data_class().get(formdata.id).evolution[-1].parts[-1].content == 'HELLO WORLD'

    add_to_journal.comment = 'HELLO WORLD 2'

    workflow.store()

    resp = get_app(pub).post(formdata.get_api_url() + 'hooks/plop/', status=200)

    assert formdef.data_class().get(formdata.id).evolution[-1].parts[-1].content == 'HELLO WORLD 2'

    # call requiring user

    add_to_journal.comment = 'HELLO WORLD 3'

    trigger.roles = ['logged-users']

    workflow.store()

    resp = get_app(pub).post(formdata.get_api_url() + 'hooks/plop/', status=403)

    resp = get_app(pub).post(sign_uri(formdata.get_api_url() + 'hooks/plop/'), status=200)

    assert formdef.data_class().get(formdata.id).evolution[-1].parts[-1].content == 'HELLO WORLD 3'

    # call requiring roles

    add_to_journal.comment = 'HELLO WORLD 4'

    trigger.roles = ['logged-users']

    workflow.store()

    Role.wipe()

    role = Role(name='xxx')

    role.store()

    trigger.roles = [role.id]

    workflow.store()

    resp = get_app(pub).post(sign_uri(formdata.get_api_url() + 'hooks/plop/'), status=403)

    resp = get_app(pub).post(sign_uri(formdata.get_api_url() + 'hooks/plop/', user=local_user), status=403)

    local_user.roles = [role.id]

    local_user.store()

    resp = get_app(pub).post(sign_uri(formdata.get_api_url() + 'hooks/plop/', user=local_user), status=200)

    assert formdef.data_class().get(formdata.id).evolution[-1].parts[-1].content == 'HELLO WORLD 4'

def test_global_webservice_trigger(pub):

    wf = Workflow(name='global actions')

    ac1 = wf.add_global_action('Action', 'ac1')

    trigger = ac1.append_trigger('webservice')

    trigger.identifier = 'plop'

    wf2 = assert_import_export_works(wf, include_id=True)

    assert wf2.global_actions[0].triggers[-1].id == trigger.id

    assert wf2.global_actions[0].triggers[-1].identifier == trigger.identifier

            ('webservice', _('Webservice')),

            raise AccessForbiddenError('user not authenticated X')

                raise AccessForbiddenError('user not authenticated Y')

                raise AccessForbiddenError('user not authenticated Z')

        # check access for all paths (except webooks), to block access to

        # formdata that would otherwise be accessible if the user is the

        # submitter.

        if not self.is_webhook:

            self.check_access()

    def _q_traverse(self, path):

        self.is_webhook = False

        if len(path) > 1:

            # webhooks have their own access checks, request cannot be blocked

            # at this point.

            self.is_webhook = bool(path[1] == 'hooks')

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

                raise AccessForbiddenError('user not authenticated W')

# w.c.s. - web application for online forms

# Copyright (C) 2005-2019  Entr'ouvert

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, see <http://www.gnu.org/licenses/>.

import json

from quixote import get_request, get_response

from quixote.directory import Directory

from qommon import errors

from wcs.api import get_user_from_api_query_string, is_url_signed

from wcs.roles import logged_users_role

from wcs.workflows import (

        WorkflowGlobalActionWebserviceTrigger,

        get_role_translation,

        perform_items)

class HookDirectory(Directory):

    _q_exports = ['']

    def __init__(self, formdata, action, trigger):

        self.formdata = formdata

        self.action = action

        self.trigger = trigger

    def _q_index(self):

        get_response().set_content_type('application/json')

        if not get_request().get_method() == 'POST':

            raise errors.AccessForbiddenError('must be POST')

        user = get_user_from_api_query_string() or get_request().user

        if self.trigger.roles:

            for role in self.trigger.roles:

                if role == logged_users_role().id and (user or is_url_signed()):

                    break

                if role == '_submitter' and self.formdata.is_submitter(user):

                    break

                if not user:

                    continue

                if get_role_translation(self.formdata, role) in (user.roles or []):

                    break

            else:

                raise errors.AccessForbiddenError('insufficient roles')

        perform_items(self.action.items, self.formdata)

        return json.dumps({'err': 0})

class WorkflowGlobalActionWebserviceHooksDirectory(Directory):

    def __init__(self, formdata):

        self.formdata = formdata

    def _q_lookup(self, component):

        for action in self.formdata.formdef.workflow.global_actions:

            for trigger in action.triggers or []:

                if isinstance(trigger, WorkflowGlobalActionWebserviceTrigger):

                    if trigger.identifier == component:

                        return HookDirectory(self.formdata, action, trigger)

        raise errors.TraversalError()

        directories = []

        for action in self.global_actions:

            for trigger in action.triggers or []:

                directories.extend(trigger.get_subdirectories(formdata))

        directories.extend(wf_status.get_subdirectories(formdata))

        return directories

    def get_subdirectories(self, formdata):

        return []

class WorkflowGlobalActionWebserviceTrigger(WorkflowGlobalActionManualTrigger):

    key = 'webservice'

    identifier = None

    roles = None

    def get_parameters(self):

        return ('identifier', 'roles')

    def render_as_line(self):

        if self.identifier:

            return _('Webservice (%s)') % self.identifier

        else:

            return _('Webservice (not configured)')

    def form(self, workflow):

        form = Form(enctype='multipart/form-data')

        form.add(StringWidget, 'identifier', title=_('Identifier'),

                required=True, value=self.identifier)

        options = [(None, '---', None)]

        options += workflow.get_list_of_roles(include_logged_in_users=True)

        form.add(WidgetList, 'roles', title=_('Roles'),

                element_type=SingleSelectWidget,

                value=self.roles,

                add_element_label=_('Add Role'),

                element_kwargs={'render_br': False,

                                'options': options})

        return form

    def get_subdirectories(self, formdata):

        from wcs.forms.workflows import WorkflowGlobalActionWebserviceHooksDirectory

        return [('hooks', WorkflowGlobalActionWebserviceHooksDirectory(formdata))]

            'timeout': WorkflowGlobalActionTimeoutTrigger,

            'webservice': WorkflowGlobalActionWebserviceTrigger,