4062021824

Build # 4062021824

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #58 from zopefoundation/config-with-zope-product-template-3b0d3566

Run Details

84 of 147 branches covered (57.14%)

Branch coverage included in aggregate %.

27 of 27 new or added lines in 4 files covered. (100.0%)

898 of 1006 relevant lines covered (89.26%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.2

/src/Products/PythonScripts/PythonScript.py

##############################################################################
#
# Copyright (c) 2002 Zope Foundation and Contributors.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Python Scripts Product

This product provides support for Script objects containing restricted
Python code.
"""

import importlib.util
import marshal
import os
import re
import sys
import types
from logging import getLogger
from urllib.parse import quote

from AccessControl.class_init import InitializeClass
from AccessControl.Permissions import change_proxy_roles
from AccessControl.Permissions import change_python_scripts
from AccessControl.Permissions import view_management_screens
from AccessControl.requestmethod import requestmethod
from AccessControl.SecurityInfo import ClassSecurityInfo
from AccessControl.SecurityManagement import getSecurityManager
from AccessControl.ZopeGuards import get_safe_globals
from AccessControl.ZopeGuards import guarded_getattr
from Acquisition import aq_parent
from App.Common import package_home
from App.special_dtml import DTMLFile
from OFS.Cache import Cacheable
from OFS.History import Historical
from OFS.History import html_diff
from OFS.SimpleItem import SimpleItem
from RestrictedPython import compile_restricted_function
from Shared.DC.Scripts.Script import BindingsUI
from Shared.DC.Scripts.Script import Script
from Shared.DC.Scripts.Script import defaultBindings
from zExceptions import Forbidden
from zExceptions import ResourceLockedError
from ZPublisher.HTTPRequest import default_encoding


LOG = getLogger('PythonScripts')

# Track the Python bytecode version
Python_magic = importlib.util.MAGIC_NUMBER

# This should only be incremented to force recompilation.
Script_magic = 4
_log_complaint = (
    'Some of your Scripts have stale code cached.  Since Zope cannot'
    ' use this code, startup will be slightly slower until these Scripts'
    ' are edited. You can automatically recompile all Scripts that have'
    ' this problem by visiting /manage_addProduct/PythonScripts/recompile'
    ' of your server in a browser.')
manage_addPythonScriptForm = DTMLFile('www/pyScriptAdd', globals())
_default_file = os.path.join(package_home(globals()), 'www', 'default_content')
_marker = []  # Create a new marker object


def manage_addPythonScript(self, id, title='', file=None, REQUEST=None,
                           submit=None):
    """Add a Python script to a folder.
    """
    id = str(id)
    id = self._setObject(id, PythonScript(id))
    pyscript = self._getOb(id)
    if title:
        pyscript.ZPythonScript_setTitle(title)

    file = file or (REQUEST and REQUEST.form.get('file'))
    if hasattr(file, 'read'):
        file = file.read()
    if not file:
        with open(_default_file) as fp:
            file = fp.read()
    pyscript.write(file)

    if REQUEST is not None:
        try:
            u = self.DestinationURL()
        except Exception:
            u = REQUEST['URL1']
        if submit == 'Add and Edit':
            u = f'{u}/{quote(id)}'
        REQUEST.RESPONSE.redirect(u + '/manage_main')
    return ''


class PythonScript(Script, Historical, Cacheable):
    """Web-callable scripts written in a safe subset of Python.

    The function may include standard python code, so long as it does
    not attempt to use the "exec" statement or certain restricted builtins.
    """

    meta_type = 'Script (Python)'
    zmi_icon = 'fa fa-terminal'
    _proxy_roles = ()

    _params = _body = ''
    errors = warnings = ()
    _v_change = 0

    manage_options = (
        {'label': 'Edit', 'action': 'ZPythonScriptHTML_editForm'},
    ) + BindingsUI.manage_options + (
        {'label': 'Test', 'action': 'ZScriptHTML_tryForm'},
        {'label': 'Proxy', 'action': 'manage_proxyForm'},
    ) + Historical.manage_options + SimpleItem.manage_options + \
        Cacheable.manage_options

    def __init__(self, id):
        self.ZBindings_edit(defaultBindings)
        bind_names = self.getBindingAssignments().getAssignedNamesInOrder()
        if id in bind_names:
            raise ValueError(
                'Following names are not allowed as identifiers, as they'
                'have a special meaning for PythonScript: '
                '%s.'
                'Please choose another name.' % ', '.join(bind_names))
        self.id = id
        self._makeFunction()

    security = ClassSecurityInfo()

    security.declareObjectProtected('View')
    security.declareProtected('View', '__call__')  # noqa: D001

    security.declareProtected(  # noqa: D001
        view_management_screens,
        'ZPythonScriptHTML_editForm', 'manage_main', 'ZScriptHTML_tryForm')

    ZPythonScriptHTML_editForm = DTMLFile('www/pyScriptEdit', globals())
    manage = manage_main = ZPythonScriptHTML_editForm
    ZPythonScriptHTML_editForm._setName('ZPythonScriptHTML_editForm')

    @security.protected(change_python_scripts)
    def ZPythonScriptHTML_editAction(self, REQUEST, title, params, body):
        """Change the script's main parameters."""
        self.ZPythonScript_setTitle(title)
        self.ZPythonScript_edit(params, body)
        message = 'Saved changes.'
        return self.ZPythonScriptHTML_editForm(self, REQUEST,
                                               manage_tabs_message=message)

    @security.protected(change_python_scripts)
    def ZPythonScript_setTitle(self, title):
        title = str(title)
        if self.title != title:
            self.title = title
            self.ZCacheable_invalidate()

    @security.protected(change_python_scripts)
    def ZPythonScript_edit(self, params, body):
        self._validateProxy()
        if self.wl_isLocked():
            raise ResourceLockedError('The script is locked via WebDAV.')
        if not isinstance(body, str):
            body = body.read()

        if self._params != params or self._body != body or self._v_change:
            self._params = str(params)
            self.write(body)

    @security.protected(change_python_scripts)
    def ZPythonScriptHTML_upload(self, REQUEST, file=''):
        """Replace the body of the script with the text in file."""
        if self.wl_isLocked():
            raise ResourceLockedError('The script is locked via WebDAV.')

        if not isinstance(file, str):
            if not file:
                return self.ZPythonScriptHTML_editForm(
                    self, REQUEST,
                    manage_tabs_message='No file specified',
                    manage_tabs_type='warning')
            file = file.read()

        self.write(file)
        message = 'Saved changes.'
        return self.ZPythonScriptHTML_editForm(self, REQUEST,
                                               manage_tabs_message=message)

    def ZScriptHTML_tryParams(self):
        """Parameters to test the script with."""
        param_names = []
        for name in self._params.split(','):

            name = name.strip()
            if name and name[0] != '*' and re.match(r'\w', name):
                param_names.append(name.split('=', 1)[0].strip())
        return param_names

    @security.protected(view_management_screens)
    def manage_historyCompare(self, rev1, rev2, REQUEST,
                              historyComparisonResults=''):
        return PythonScript.inheritedAttribute('manage_historyCompare')(
            self, rev1, rev2, REQUEST,
            historyComparisonResults=html_diff(rev1.read(), rev2.read()))

    def __setstate__(self, state):
        Script.__setstate__(self, state)
        if getattr(self, 'Python_magic', None) != Python_magic or \
           getattr(self, 'Script_magic', None) != Script_magic:
            global _log_complaint
            if _log_complaint:
                LOG.info(_log_complaint)
                _log_complaint = 0
            # Changes here won't get saved, unless this Script is edited.
            body = self._body.rstrip()
            if body:
                self._body = body + '\n'
            self._compile()
            self._v_change = 1
        elif self._code is None:
            self._v_ft = None
        else:
            self._newfun(marshal.loads(self._code))

    def _compile(self):
        bind_names = self.getBindingAssignments().getAssignedNamesInOrder()
        compile_result = compile_restricted_function(
            self._params,
            body=self._body or 'pass',
            name=self.id,
            filename=self.meta_type,
            globalize=bind_names)

        code = compile_result.code
        errors = compile_result.errors
        self.warnings = tuple(compile_result.warnings)
        if errors:
            self._code = None
            self._v_ft = None
            self._setFuncSignature((), (), 0)
            # Fix up syntax errors.
            filestring = '  File "<string>",'
            for i in range(len(errors)):
                line = errors[i]
                if line.startswith(filestring):
                    errors[i] = line.replace(filestring, '  Script', 1)
            self.errors = errors
            return

        self._code = marshal.dumps(code)
        self.errors = ()
        f = self._newfun(code)
        fc = f.__code__
        self._setFuncSignature(f.__defaults__, fc.co_varnames,
                               fc.co_argcount)
        self.Python_magic = Python_magic
        self.Script_magic = Script_magic
        self._v_change = 0

    def _newfun(self, code):
        safe_globals = get_safe_globals()
        safe_globals['_getattr_'] = guarded_getattr
        safe_globals['__debug__'] = __debug__
        # it doesn't really matter what __name__ is, *but*
        # - we need a __name__
        #   (see testPythonScript.TestPythonScriptGlobals.test__name__)
        # - it should not contain a period, so we can't use the id
        #   (see https://bugs.launchpad.net/zope2/+bug/142731/comments/4)
        # - with Python 2.6 it should not be None
        #   (see testPythonScript.TestPythonScriptGlobals.test_filepath)
        safe_globals['__name__'] = 'script'

        safe_locals = {}
        exec(code, safe_globals, safe_locals)
        func = list(safe_locals.values())[0]
        self._v_ft = (func.__code__, safe_globals, func.__defaults__ or ())
        return func

    def _makeFunction(self):
        self.ZCacheable_invalidate()
        self._compile()
        if not (aq_parent(self) is None or hasattr(self, '_filepath')):
            # It needs a _filepath, and has an acquisition wrapper.
            self._filepath = self.get_filepath()

    def _editedBindings(self):
        if getattr(self, '_v_ft', None) is not None:
            self._makeFunction()

    def _exec(self, bound_names, args, kw):
        """Call a Python Script

        Calling a Python Script is an actual function invocation.
        """
        # Retrieve the value from the cache.
        keyset = None
        if self.ZCacheable_isCachingEnabled():
            # Prepare a cache key.
            keyset = kw.copy()
            asgns = self.getBindingAssignments()
            name_context = asgns.getAssignedName('name_context', None)
            if name_context:
                keyset[name_context] = aq_parent(self).getPhysicalPath()
            name_subpath = asgns.getAssignedName('name_subpath', None)
            if name_subpath:
                keyset[name_subpath] = self._getTraverseSubpath()
            # Note: perhaps we should cache based on name_ns also.
            keyset['*'] = args
            result = self.ZCacheable_get(keywords=keyset, default=_marker)
            if result is not _marker:
                # Got a cached value.
                return result

        ft = self._v_ft
        if ft is None:
            __traceback_supplement__ = (
                PythonScriptTracebackSupplement, self)
            raise RuntimeError(f'{self.meta_type} {self.id} has errors.')

        function_code, safe_globals, function_argument_definitions = ft
        safe_globals = safe_globals.copy()
        if bound_names is not None:
            safe_globals.update(bound_names)
        safe_globals['__traceback_supplement__'] = (
            PythonScriptTracebackSupplement, self, -1)
        safe_globals['__file__'] = getattr(
            self, '_filepath', None) or self.get_filepath()
        function = types.FunctionType(
            function_code, safe_globals, None, function_argument_definitions)

        try:
            result = function(*args, **kw)
        except SystemExit:
            raise ValueError(
                'SystemExit cannot be raised within a PythonScript')

        if keyset is not None:
            # Store the result in the cache.
            self.ZCacheable_set(result, keywords=keyset)
        return result

    def manage_afterAdd(self, item, container):
        if item is self:
            self._filepath = self.get_filepath()

    def manage_beforeDelete(self, item, container):
        # shut up deprecation warnings
        pass

    def manage_afterClone(self, item):
        # shut up deprecation warnings
        pass

    @security.protected(view_management_screens)
    def get_filepath(self):
        return self.meta_type + ':' + '/'.join(self.getPhysicalPath())

    def manage_haveProxy(self, r):
        return r in self._proxy_roles

    def _validateProxy(self, roles=None):
        if roles is None:
            roles = self._proxy_roles
        if not roles:
            return
        user = getSecurityManager().getUser()
        if user is not None and user.allowed(self, roles):
            return
        raise Forbidden(
            'You are not authorized to change <em>%s</em> '
            'because you do not have proxy roles.\n<!--%s, %s-->' % (
                self.id, user, roles))

    security.declareProtected(change_proxy_roles,  # NOQA: D001
                              'manage_proxyForm')

    manage_proxyForm = DTMLFile('www/pyScriptProxy', globals())

    @security.protected(change_proxy_roles)
    @requestmethod('POST')
    def manage_proxy(self, roles=(), REQUEST=None):
        """Change Proxy Roles"""
        user = getSecurityManager().getUser()
        if 'Manager' not in user.getRolesInContext(self):
            self._validateProxy(roles)
            self._validateProxy()
        self.ZCacheable_invalidate()
        self._proxy_roles = tuple(roles)
        if REQUEST:
            msg = 'Proxy roles changed.'
            return self.manage_proxyForm(manage_tabs_message=msg,
                                         management_view='Proxy')

    security.declareProtected(  # NOQA: D001
        change_python_scripts,
        'manage_FTPput', 'manage_historyCopy',
        'manage_beforeHistoryCopy', 'manage_afterHistoryCopy')

    @security.protected(change_python_scripts)
    def PUT(self, REQUEST, RESPONSE):
        """ Handle HTTP PUT requests """
        self.dav__init(REQUEST, RESPONSE)
        self.dav__simpleifhandler(REQUEST, RESPONSE, refresh=1)
        new_body = REQUEST.get('BODY', '')
        self.write(new_body)
        RESPONSE.setStatus(204)
        return RESPONSE

    manage_FTPput = PUT

    @security.protected(change_python_scripts)
    def write(self, text):
        """ Change the Script by parsing a read()-style source text. """
        self._validateProxy()
        mdata = self._metadata_map()
        bindmap = self.getBindingAssignments().getAssignedNames()
        bup = 0

        if isinstance(text, bytes):
            text = text.decode(default_encoding)

        st = 0
        try:
            while 1:
                # Find the next non-empty line
                m = _nonempty_line.search(text, st)
                if not m:
                    # There were no non-empty body lines
                    body = ''
                    break
                line = m.group(0).strip()
                if line[:2] != '##':
                    # We have found the first line of the body
                    body = text[m.start(0):]
                    break

                st = m.end(0)
                # Parse this header line
                if len(line) == 2 or line[2] == ' ' or '=' not in line:
                    # Null header line
                    continue
                k, v = line[2:].split('=', 1)
                k = k.strip().lower()
                v = v.strip()
                if k not in mdata:
                    raise SyntaxError('Unrecognized header line "%s"' % line)
                if v == mdata[k]:
                    # Unchanged value
                    continue

                # Set metadata value
                if k == 'title':
                    self.title = v
                elif k == 'parameters':
                    self._params = v
                elif k[:5] == 'bind ':
                    bindmap[_nice_bind_names[k[5:]]] = v
                    bup = 1

            body = body.rstrip()
            if body:
                body = body + '\n'
            if body != self._body:
                self._body = body
            if bup:
                self.ZBindings_edit(bindmap)
            else:
                self._makeFunction()
        except Exception:
            LOG.error('write failed', exc_info=sys.exc_info())
            raise

    def manage_DAVget(self):
        """Get source for WebDAV"""
        self.REQUEST.RESPONSE.setHeader('Content-Type', 'text/plain')
        return self.read()

    manage_FTPget = manage_DAVget

    def _metadata_map(self):
        m = {
            'title': self.title,
            'parameters': self._params,
        }
        bindmap = self.getBindingAssignments().getAssignedNames()
        for k, v in _nice_bind_names.items():
            m['bind ' + k] = bindmap.get(v, '')
        return m

    @security.protected(view_management_screens)
    def read(self):
        """ Generate a text representation of the Script source.

        Includes specially formatted comment lines for parameters,
        bindings, and the title.
        """
        # Construct metadata header lines, indented the same as the body.
        m = _first_indent.search(self._body)
        if m:
            prefix = m.group(0) + '##'
        else:
            prefix = '##'

        hlines = [f'{prefix} {self.meta_type} "{self.id}"']
        mm = sorted(self._metadata_map().items())
        for kv in mm:
            hlines.append('%s=%s' % kv)
        if self.errors:
            hlines.append('')
            hlines.append(' Errors:')
            for line in self.errors:
                hlines.append('  ' + line)
        if self.warnings:
            hlines.append('')
            hlines.append(' Warnings:')
            for line in self.warnings:
                hlines.append('  ' + line)
        hlines.append('')
        return ('\n' + prefix).join(hlines) + '\n' + self._body

    @security.protected(view_management_screens)
    def params(self):
        return self._params

    @security.protected(view_management_screens)
    def body(self):
        return self._body

    def get_size(self):
        return len(self.read())

    getSize = get_size

    @security.protected(view_management_screens)
    def PrincipiaSearchSource(self):
        """Support for searching - the document's contents are searched."""
        return f'{self._params}\n{self._body}'

    @security.protected(view_management_screens)
    def document_src(self, REQUEST=None, RESPONSE=None):
        """Return unprocessed document source."""

        if RESPONSE is not None:
            RESPONSE.setHeader('Content-Type', 'text/plain')
        return self.read()


InitializeClass(PythonScript)


class PythonScriptTracebackSupplement:
    """Implementation of ITracebackSupplement"""
    def __init__(self, script, line=0):
        self.object = script
        # If line is set to -1, it means to use tb_lineno.
        self.line = line


_first_indent = re.compile('(?m)^ *(?! |$)')
_nonempty_line = re.compile(r'(?m)^(.*\S.*)$')

_nice_bind_names = {'context': 'name_context', 'container': 'name_container',
                    'script': 'name_m_self', 'namespace': 'name_ns',
                    'subpath': 'name_subpath'}

1	##############################################################################
2	#
3	# Copyright (c) 2002 Zope Foundation and Contributors.
4	#
5	# This software is subject to the provisions of the Zope Public License,
6	# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
7	# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
8	# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
9	# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
10	# FOR A PARTICULAR PURPOSE.
11	#
12	##############################################################################
13	"""Python Scripts Product	1✔
14
15	This product provides support for Script objects containing restricted
16	Python code.
17	"""
18
19	import importlib.util	1✔
20	import marshal	1✔
21	import os	1✔
22	import re	1✔
23	import sys	1✔
24	import types	1✔
25	from logging import getLogger	1✔
26	from urllib.parse import quote	1✔
27
28	from AccessControl.class_init import InitializeClass	1✔
29	from AccessControl.Permissions import change_proxy_roles	1✔
30	from AccessControl.Permissions import change_python_scripts	1✔
31	from AccessControl.Permissions import view_management_screens	1✔
32	from AccessControl.requestmethod import requestmethod	1✔
33	from AccessControl.SecurityInfo import ClassSecurityInfo	1✔
34	from AccessControl.SecurityManagement import getSecurityManager	1✔
35	from AccessControl.ZopeGuards import get_safe_globals	1✔
36	from AccessControl.ZopeGuards import guarded_getattr	1✔
37	from Acquisition import aq_parent	1✔
38	from App.Common import package_home	1✔
39	from App.special_dtml import DTMLFile	1✔
40	from OFS.Cache import Cacheable	1✔
41	from OFS.History import Historical	1✔
42	from OFS.History import html_diff	1✔
43	from OFS.SimpleItem import SimpleItem	1✔
44	from RestrictedPython import compile_restricted_function	1✔
45	from Shared.DC.Scripts.Script import BindingsUI	1✔
46	from Shared.DC.Scripts.Script import Script	1✔
47	from Shared.DC.Scripts.Script import defaultBindings	1✔
48	from zExceptions import Forbidden	1✔
49	from zExceptions import ResourceLockedError	1✔
50	from ZPublisher.HTTPRequest import default_encoding	1✔
51
52
53	LOG = getLogger('PythonScripts')	1✔
54
55	# Track the Python bytecode version
56	Python_magic = importlib.util.MAGIC_NUMBER	1✔
57
58	# This should only be incremented to force recompilation.
59	Script_magic = 4	1✔
60	_log_complaint = (	1✔
61	'Some of your Scripts have stale code cached. Since Zope cannot'
62	' use this code, startup will be slightly slower until these Scripts'
63	' are edited. You can automatically recompile all Scripts that have'
64	' this problem by visiting /manage_addProduct/PythonScripts/recompile'
65	' of your server in a browser.')
66	manage_addPythonScriptForm = DTMLFile('www/pyScriptAdd', globals())	1✔
67	_default_file = os.path.join(package_home(globals()), 'www', 'default_content')	1✔
68	_marker = [] # Create a new marker object	1✔
69
70
71	def manage_addPythonScript(self, id, title='', file=None, REQUEST=None,	1✔
72	submit=None):
73	"""Add a Python script to a folder.
74	"""
75	id = str(id)	1✔
76	id = self._setObject(id, PythonScript(id))	1✔
77	pyscript = self._getOb(id)	1✔
78	if title:	1✔
79	pyscript.ZPythonScript_setTitle(title)	1✔
80
81	file = file or (REQUEST and REQUEST.form.get('file'))	1✔
82	if hasattr(file, 'read'):	1✔
83	file = file.read()	1✔
84	if not file:	1✔
85	with open(_default_file) as fp:	1✔
86	file = fp.read()	1✔
87	pyscript.write(file)	1✔
88
89	if REQUEST is not None:	1✔
90	try:	1✔
91	u = self.DestinationURL()	1✔
92	except Exception:	1✔
93	u = REQUEST['URL1']	1✔
94	if submit == 'Add and Edit':	1!
95	u = f'{u}/{quote(id)}'	×
96	REQUEST.RESPONSE.redirect(u + '/manage_main')	1✔
97	return ''	1✔
98
99
100	class PythonScript(Script, Historical, Cacheable):	1✔
101	"""Web-callable scripts written in a safe subset of Python.
102
103	The function may include standard python code, so long as it does
104	not attempt to use the "exec" statement or certain restricted builtins.
105	"""
106
107	meta_type = 'Script (Python)'	1✔
108	zmi_icon = 'fa fa-terminal'	1✔
109	_proxy_roles = ()	1✔
110
111	_params = _body = ''	1✔
112	errors = warnings = ()	1✔
113	_v_change = 0	1✔
114
115	manage_options = (	1✔
116	{'label': 'Edit', 'action': 'ZPythonScriptHTML_editForm'},
117	) + BindingsUI.manage_options + (
118	{'label': 'Test', 'action': 'ZScriptHTML_tryForm'},
119	{'label': 'Proxy', 'action': 'manage_proxyForm'},
120	) + Historical.manage_options + SimpleItem.manage_options + \
121	Cacheable.manage_options
122
123	def __init__(self, id):	1✔
124	self.ZBindings_edit(defaultBindings)	1✔
125	bind_names = self.getBindingAssignments().getAssignedNamesInOrder()	1✔
126	if id in bind_names:	1✔
127	raise ValueError(	1✔
128	'Following names are not allowed as identifiers, as they'
129	'have a special meaning for PythonScript: '
130	'%s.'
131	'Please choose another name.' % ', '.join(bind_names))
132	self.id = id	1✔
133	self._makeFunction()	1✔
134
135	security = ClassSecurityInfo()	1✔
136
137	security.declareObjectProtected('View')	1✔
138	security.declareProtected('View', '__call__') # noqa: D001	1✔
139
140	security.declareProtected( # noqa: D001	1✔
141	view_management_screens,
142	'ZPythonScriptHTML_editForm', 'manage_main', 'ZScriptHTML_tryForm')
143
144	ZPythonScriptHTML_editForm = DTMLFile('www/pyScriptEdit', globals())	1✔
145	manage = manage_main = ZPythonScriptHTML_editForm	1✔
146	ZPythonScriptHTML_editForm._setName('ZPythonScriptHTML_editForm')	1✔
147
148	@security.protected(change_python_scripts)	1✔
149	def ZPythonScriptHTML_editAction(self, REQUEST, title, params, body):	1✔
150	"""Change the script's main parameters."""
151	self.ZPythonScript_setTitle(title)	×
152	self.ZPythonScript_edit(params, body)	×
153	message = 'Saved changes.'	×
154	return self.ZPythonScriptHTML_editForm(self, REQUEST,	×
155	manage_tabs_message=message)
156
157	@security.protected(change_python_scripts)	1✔
158	def ZPythonScript_setTitle(self, title):	1✔
159	title = str(title)	1✔
160	if self.title != title:	1!
161	self.title = title	1✔
162	self.ZCacheable_invalidate()	1✔
163
164	@security.protected(change_python_scripts)	1✔
165	def ZPythonScript_edit(self, params, body):	1✔
166	self._validateProxy()	×
167	if self.wl_isLocked():	×
168	raise ResourceLockedError('The script is locked via WebDAV.')	×
169	if not isinstance(body, str):	×
170	body = body.read()	×
171
172	if self._params != params or self._body != body or self._v_change:	×
173	self._params = str(params)	×
174	self.write(body)	×
175
176	@security.protected(change_python_scripts)	1✔
177	def ZPythonScriptHTML_upload(self, REQUEST, file=''):	1✔
178	"""Replace the body of the script with the text in file."""
179	if self.wl_isLocked():	1!
180	raise ResourceLockedError('The script is locked via WebDAV.')	×
181
182	if not isinstance(file, str):	1!
183	if not file:	1✔
184	return self.ZPythonScriptHTML_editForm(	1✔
185	self, REQUEST,
186	manage_tabs_message='No file specified',
187	manage_tabs_type='warning')
188	file = file.read()	1✔
189
190	self.write(file)	1✔
191	message = 'Saved changes.'	1✔
192	return self.ZPythonScriptHTML_editForm(self, REQUEST,	1✔
193	manage_tabs_message=message)
194
195	def ZScriptHTML_tryParams(self):	1✔
196	"""Parameters to test the script with."""
197	param_names = []	×
198	for name in self._params.split(','):	×
199
200	name = name.strip()	×
201	if name and name[0] != '*' and re.match(r'\w', name):	×
202	param_names.append(name.split('=', 1)[0].strip())	×
203	return param_names	×
204
205	@security.protected(view_management_screens)	1✔
206	def manage_historyCompare(self, rev1, rev2, REQUEST,	1✔
207	historyComparisonResults=''):
208	return PythonScript.inheritedAttribute('manage_historyCompare')(	×
209	self, rev1, rev2, REQUEST,
210	historyComparisonResults=html_diff(rev1.read(), rev2.read()))
211
212	def __setstate__(self, state):	1✔
213	Script.__setstate__(self, state)	×
214	if getattr(self, 'Python_magic', None) != Python_magic or \	×
215	getattr(self, 'Script_magic', None) != Script_magic:
216	global _log_complaint
217	if _log_complaint:	×
218	LOG.info(_log_complaint)	×
219	_log_complaint = 0	×
220	# Changes here won't get saved, unless this Script is edited.
221	body = self._body.rstrip()	×
222	if body:	×
223	self._body = body + '\n'	×
224	self._compile()	×
225	self._v_change = 1	×
226	elif self._code is None:	×
227	self._v_ft = None	×
228	else:
229	self._newfun(marshal.loads(self._code))	×
230
231	def _compile(self):	1✔
232	bind_names = self.getBindingAssignments().getAssignedNamesInOrder()	1✔
233	compile_result = compile_restricted_function(	1✔
234	self._params,
235	body=self._body or 'pass',
236	name=self.id,
237	filename=self.meta_type,
238	globalize=bind_names)
239
240	code = compile_result.code	1✔
241	errors = compile_result.errors	1✔
242	self.warnings = tuple(compile_result.warnings)	1✔
243	if errors:	1✔
244	self._code = None	1✔
245	self._v_ft = None	1✔
246	self._setFuncSignature((), (), 0)	1✔
247	# Fix up syntax errors.
248	filestring = ' File "<string>",'	1✔
249	for i in range(len(errors)):	1✔
250	line = errors[i]	1✔
251	if line.startswith(filestring):	1!
252	errors[i] = line.replace(filestring, ' Script', 1)	×
253	self.errors = errors	1✔
254	return	1✔
255
256	self._code = marshal.dumps(code)	1✔
257	self.errors = ()	1✔
258	f = self._newfun(code)	1✔
259	fc = f.__code__	1✔
260	self._setFuncSignature(f.__defaults__, fc.co_varnames,	1✔
261	fc.co_argcount)
262	self.Python_magic = Python_magic	1✔
263	self.Script_magic = Script_magic	1✔
264	self._v_change = 0	1✔
265
266	def _newfun(self, code):	1✔
267	safe_globals = get_safe_globals()	1✔
268	safe_globals['_getattr_'] = guarded_getattr	1✔
269	safe_globals['__debug__'] = __debug__	1✔
270	# it doesn't really matter what __name__ is, but
271	# - we need a __name__
272	# (see testPythonScript.TestPythonScriptGlobals.test__name__)
273	# - it should not contain a period, so we can't use the id
274	# (see https://bugs.launchpad.net/zope2/+bug/142731/comments/4)
275	# - with Python 2.6 it should not be None
276	# (see testPythonScript.TestPythonScriptGlobals.test_filepath)
277	safe_globals['__name__'] = 'script'	1✔
278
279	safe_locals = {}	1✔
280	exec(code, safe_globals, safe_locals)	1✔
281	func = list(safe_locals.values())[0]	1✔
282	self._v_ft = (func.__code__, safe_globals, func.__defaults__ or ())	1✔
283	return func	1✔
284
285	def _makeFunction(self):	1✔
286	self.ZCacheable_invalidate()	1✔
287	self._compile()	1✔
288	if not (aq_parent(self) is None or hasattr(self, '_filepath')):	1✔
289	# It needs a _filepath, and has an acquisition wrapper.
290	self._filepath = self.get_filepath()	1✔
291
292	def _editedBindings(self):	1✔
293	if getattr(self, '_v_ft', None) is not None:	1✔
294	self._makeFunction()	1✔
295
296	def _exec(self, bound_names, args, kw):	1✔
297	"""Call a Python Script
298
299	Calling a Python Script is an actual function invocation.
300	"""
301	# Retrieve the value from the cache.
302	keyset = None	1✔
303	if self.ZCacheable_isCachingEnabled():	1!
304	# Prepare a cache key.
305	keyset = kw.copy()	×
306	asgns = self.getBindingAssignments()	×
307	name_context = asgns.getAssignedName('name_context', None)	×
308	if name_context:	×
309	keyset[name_context] = aq_parent(self).getPhysicalPath()	×
310	name_subpath = asgns.getAssignedName('name_subpath', None)	×
311	if name_subpath:	×
312	keyset[name_subpath] = self._getTraverseSubpath()	×
313	# Note: perhaps we should cache based on name_ns also.
314	keyset['*'] = args	×
315	result = self.ZCacheable_get(keywords=keyset, default=_marker)	×
316	if result is not _marker:	×
317	# Got a cached value.
318	return result	×
319
320	ft = self._v_ft	1✔
321	if ft is None:	1!
322	__traceback_supplement__ = (	×
323	PythonScriptTracebackSupplement, self)
324	raise RuntimeError(f'{self.meta_type} {self.id} has errors.')	×
325
326	function_code, safe_globals, function_argument_definitions = ft	1✔
327	safe_globals = safe_globals.copy()	1✔
328	if bound_names is not None:	1!
329	safe_globals.update(bound_names)	1✔
330	safe_globals['__traceback_supplement__'] = (	1✔
331	PythonScriptTracebackSupplement, self, -1)
332	safe_globals['__file__'] = getattr(	1✔
333	self, '_filepath', None) or self.get_filepath()
334	function = types.FunctionType(	1✔
335	function_code, safe_globals, None, function_argument_definitions)
336
337	try:	1✔
338	result = function(args, *kw)	1✔
339	except SystemExit:	1✔
340	raise ValueError(	1✔
341	'SystemExit cannot be raised within a PythonScript')
342
343	if keyset is not None:	1!
344	# Store the result in the cache.
345	self.ZCacheable_set(result, keywords=keyset)	×
346	return result	1✔
347
348	def manage_afterAdd(self, item, container):	1✔
349	if item is self:	1!
350	self._filepath = self.get_filepath()	1✔
351
352	def manage_beforeDelete(self, item, container):	1✔
353	# shut up deprecation warnings
354	pass	×
355
356	def manage_afterClone(self, item):	1✔
357	# shut up deprecation warnings
358	pass	×
359
360	@security.protected(view_management_screens)	1✔
361	def get_filepath(self):	1✔
362	return self.meta_type + ':' + '/'.join(self.getPhysicalPath())	1✔
363
364	def manage_haveProxy(self, r):	1✔
365	return r in self._proxy_roles	1✔
366
367	def _validateProxy(self, roles=None):	1✔
368	if roles is None:	1✔
369	roles = self._proxy_roles	1✔
370	if not roles:	1✔
371	return	1✔
372	user = getSecurityManager().getUser()	1✔
373	if user is not None and user.allowed(self, roles):	1✔
374	return	1✔
375	raise Forbidden(	1✔
376	'You are not authorized to change <em>%s</em> '
377	'because you do not have proxy roles.\n<!--%s, %s-->' % (
378	self.id, user, roles))
379
380	security.declareProtected(change_proxy_roles, # NOQA: D001	1✔
381	'manage_proxyForm')
382
383	manage_proxyForm = DTMLFile('www/pyScriptProxy', globals())	1✔
384
385	@security.protected(change_proxy_roles)	1✔
386	@requestmethod('POST')	1✔
387	def manage_proxy(self, roles=(), REQUEST=None):	1✔
388	"""Change Proxy Roles"""
389	user = getSecurityManager().getUser()	1✔
390	if 'Manager' not in user.getRolesInContext(self):	1✔
391	self._validateProxy(roles)	1✔
392	self._validateProxy()	1✔
393	self.ZCacheable_invalidate()	1✔
394	self._proxy_roles = tuple(roles)	1✔
395	if REQUEST:	1!
396	msg = 'Proxy roles changed.'	1✔
397	return self.manage_proxyForm(manage_tabs_message=msg,	1✔
398	management_view='Proxy')
399
400	security.declareProtected( # NOQA: D001	1✔
401	change_python_scripts,
402	'manage_FTPput', 'manage_historyCopy',
403	'manage_beforeHistoryCopy', 'manage_afterHistoryCopy')
404
405	@security.protected(change_python_scripts)	1✔
406	def PUT(self, REQUEST, RESPONSE):	1✔
407	""" Handle HTTP PUT requests """
408	self.dav__init(REQUEST, RESPONSE)	1✔
409	self.dav__simpleifhandler(REQUEST, RESPONSE, refresh=1)	1✔
410	new_body = REQUEST.get('BODY', '')	1✔
411	self.write(new_body)	1✔
412	RESPONSE.setStatus(204)	1✔
413	return RESPONSE	1✔
414
415	manage_FTPput = PUT	1✔
416
417	@security.protected(change_python_scripts)	1✔
418	def write(self, text):	1✔
419	""" Change the Script by parsing a read()-style source text. """
420	self._validateProxy()	1✔
421	mdata = self._metadata_map()	1✔
422	bindmap = self.getBindingAssignments().getAssignedNames()	1✔
423	bup = 0	1✔
424
425	if isinstance(text, bytes):	1✔
426	text = text.decode(default_encoding)	1✔
427
428	st = 0	1✔
429	try:	1✔
430	while 1:
431	# Find the next non-empty line
432	m = _nonempty_line.search(text, st)	1✔
433	if not m:	1✔
434	# There were no non-empty body lines
435	body = ''	1✔
436	break	1✔
437	line = m.group(0).strip()	1✔
438	if line[:2] != '##':	1✔
439	# We have found the first line of the body
440	body = text[m.start(0):]	1✔
441	break	1✔
442
443	st = m.end(0)	1✔
444	# Parse this header line
445	if len(line) == 2 or line[2] == ' ' or '=' not in line:	1!
446	# Null header line
447	continue	×
448	k, v = line[2:].split('=', 1)	1✔
449	k = k.strip().lower()	1✔
450	v = v.strip()	1✔
451	if k not in mdata:	1✔
452	raise SyntaxError('Unrecognized header line "%s"' % line)	1✔
453	if v == mdata[k]:	1✔
454	# Unchanged value
455	continue	1✔
456
457	# Set metadata value
458	if k == 'title':	1✔
459	self.title = v	1✔
460	elif k == 'parameters':	1✔
461	self._params = v	1✔
462	elif k[:5] == 'bind ':	1!
463	bindmap[_nice_bind_names[k[5:]]] = v	1✔
464	bup = 1	1✔
465
466	body = body.rstrip()	1✔
467	if body:	1✔
468	body = body + '\n'	1✔
469	if body != self._body:	1✔
470	self._body = body	1✔
471	if bup:	1✔
472	self.ZBindings_edit(bindmap)	1✔
473	else:
474	self._makeFunction()	1✔
475	except Exception:	1✔
476	LOG.error('write failed', exc_info=sys.exc_info())	1✔
477	raise	1✔
478
479	def manage_DAVget(self):	1✔
480	"""Get source for WebDAV"""
481	self.REQUEST.RESPONSE.setHeader('Content-Type', 'text/plain')	1✔
482	return self.read()	1✔
483
484	manage_FTPget = manage_DAVget	1✔
485
486	def _metadata_map(self):	1✔
487	m = {	1✔
488	'title': self.title,
489	'parameters': self._params,
490	}
491	bindmap = self.getBindingAssignments().getAssignedNames()	1✔
492	for k, v in _nice_bind_names.items():	1✔
493	m['bind ' + k] = bindmap.get(v, '')	1✔
494	return m	1✔
495
496	@security.protected(view_management_screens)	1✔
497	def read(self):	1✔
498	""" Generate a text representation of the Script source.
499
500	Includes specially formatted comment lines for parameters,
501	bindings, and the title.
502	"""
503	# Construct metadata header lines, indented the same as the body.
504	m = _first_indent.search(self._body)	1✔
505	if m:	1!
506	prefix = m.group(0) + '##'	1✔
507	else:
508	prefix = '##'	×
509
510	hlines = [f'{prefix} {self.meta_type} "{self.id}"']	1✔
511	mm = sorted(self._metadata_map().items())	1✔
512	for kv in mm:	1✔
513	hlines.append('%s=%s' % kv)	1✔
514	if self.errors:	1!
515	hlines.append('')	×
516	hlines.append(' Errors:')	×
517	for line in self.errors:	×
518	hlines.append(' ' + line)	×
519	if self.warnings:	1!
520	hlines.append('')	×
521	hlines.append(' Warnings:')	×
522	for line in self.warnings:	×
523	hlines.append(' ' + line)	×
524	hlines.append('')	1✔
525	return ('\n' + prefix).join(hlines) + '\n' + self._body	1✔
526
527	@security.protected(view_management_screens)	1✔
528	def params(self):	1✔
529	return self._params	1✔
530
531	@security.protected(view_management_screens)	1✔
532	def body(self):	1✔
533	return self._body	1✔
534
535	def get_size(self):	1✔
536	return len(self.read())	1✔
537
538	getSize = get_size	1✔
539
540	@security.protected(view_management_screens)	1✔
541	def PrincipiaSearchSource(self):	1✔
542	"""Support for searching - the document's contents are searched."""
543	return f'{self._params}\n{self._body}'	×
544
545	@security.protected(view_management_screens)	1✔
546	def document_src(self, REQUEST=None, RESPONSE=None):	1✔
547	"""Return unprocessed document source."""
548
549	if RESPONSE is not None:	×
550	RESPONSE.setHeader('Content-Type', 'text/plain')	×
551	return self.read()	×
552
553
554	InitializeClass(PythonScript)	1✔
555
556
557	class PythonScriptTracebackSupplement:	1✔
558	"""Implementation of ITracebackSupplement"""
559	def __init__(self, script, line=0):	1✔
560	self.object = script	×
561	# If line is set to -1, it means to use tb_lineno.
562	self.line = line	×
563
564
565	_first_indent = re.compile('(?m)^ *(?! \|$)')	1✔
566	_nonempty_line = re.compile(r'(?m)^(.\S.)$')	1✔
567
568	_nice_bind_names = {'context': 'name_context', 'container': 'name_container',	1✔
569	'script': 'name_m_self', 'namespace': 'name_ns',
570	'subpath': 'name_subpath'}

zopefoundation / Products.PythonScripts / 4062021824

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous