16248875199

Committed 17 Mar 2025 07:48AM CUT coverage: 86.279%. Remained the same

Build # 16248875199

Build Type

push

github

Committed by

web-flow

Commit Message

Update Python version support. (#68)

* Drop support for Python 3.8.

* Configuring for zope-product

* Update Python version support.

Run Details

84 of 144 branches covered (58.33%)

Branch coverage included in aggregate %.

941 of 1044 relevant lines covered (90.13%)

0.9 hits per line

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

77.62

/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.abc
import importlib.util
import linecache
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 = 5
_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 PythonScriptLoader(importlib.abc.Loader):
    """PEP302 loader to display source code in tracebacks
    """

    def __init__(self, source):
        self._source = source

    def get_source(self, name):
        return self._source


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 file:
            return self.ZPythonScriptHTML_editForm(
                self, REQUEST,
                manage_tabs_message='No file specified',
                manage_tabs_type='warning')
        if not isinstance(file, str):
            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=getattr(self, '_filepath', None) or self.get_filepath(),
            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
        linecache.clearcache()
        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()
        safe_globals['__loader__'] = PythonScriptLoader(self._body)

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

zopefoundation / Products.PythonScripts / 16248875199

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