#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# TOMUSS: The Online Multi User Simple Spreadsheet
# Copyright (C) 2008-2012 Thierry EXCOFFIER, Universite Claude Bernard
#
# 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, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
# Contact: Thierry.EXCOFFIER@bat710.univ-lyon1.fr
"""
Plugin management
"""
import codecs
import re
import html
import os
import sys
import time
import weakref
import gc
from . import configuration
from . import utilities
from . import files
from . import authentication
class Link: # pylint: disable=too-many-instance-attributes
"""Define a link to be inserted on the home page Actions"""
def __init__(self,
text=None,
help='', # pylint: disable=redefined-builtin
url=None,
target="_blank",
html_class='leaves',
priority=0,
where=None,
group="",
key=None, # See TEMPLATES/config_home.py
# DEPRECATED, use 'group'
authorized=None
): # pylint: disable=too-many-arguments
if text is None:
text = utilities._('LINK_' + (url or ''))
self.text = text
if not help:
help = utilities._('HELP_' + (url or ''))
self.help = re.sub('[ \t\n]+', ' ', help)
self.url = url
self.target = target
self.html_class = html_class
self.priority = priority
self.where = where
self.plugin = None
self.group = group
self.key = key
if authorized: # pragma: no cover
utilities.warn("'authorized=' is DEPRECATED, use 'group=' in place for '%s'"%
self.text, what='Warning')
else:
authorized = lambda server: True
self.authorized = authorized
# Where the plugin is defined
self.module = sys._getframe(1).f_code.co_filename
def __str__(self):
return 'Link(%s,%s)' % (self.text, self.url) # pragma: no cover
def get_url_and_target(self, plugin):
"""Returns the URL of the link"""
if self.target:
target = ' target="' + self.target + '"'
else:
target = '' # pragma: no cover
url = self.url
if url is None:
if len(self.plugin().url) != 1:
url = ("javascript:alert('" + self.plugin().name # pragma: no cover
+ ": you need to indicate «url=» in the Link() definition.')"
)
else:
url = '/' + plugin.url[0]
if url.startswith('javascript:'):
target = ''
elif url.startswith('/'):
target = ''
return url, target
def set_plugin(self, plugin):
"""Set the link text and help from plugin"""
self.plugin = weakref.ref(plugin)
if self.text.startswith('LINK_'):
self.text = utilities._('LINK_' + plugin.name)
if self.help.startswith('HELP_'):
self.help = utilities._('HELP_' + plugin.name)
# Helper functions for parsing URL.
# They return:
# False : if it does not match
# None : if it matches and the parsing continue
# list : it matches and the parsing stops. the_path will be set to the list
def _year(server, path, i):
server.the_year = int(path[i])
def _time(server, path, i):
server.the_time = int(path[i])
def _something(server, path, i):
server.something = path[i]
def _anything(_server, path, i):
return path[i:]
def _semester(server, path, i):
server.the_semester = utilities.safe(path[i]).replace('.', '_')
def _ue(server, path, i):
server.the_ue = utilities.safe(path[i]).replace('.', '_')
def _page(server, path, i):
server.the_page = int(path[i])
def _int(server, path, i):
if not configuration.is_a_student(path[i]):
return False
server.student = configuration.User(path[i])
server.the_student = server.student.ID_
return None
def __int(server, path, i):
if not path[i].startswith('_'):
return False
server.student = configuration.User(path[i][1:])
server.the_student = server.student.ID_
return None
def _request_number(server, path, i):
path = path[i]
if not path:
return False # pragma: no cover
if not path[0].isdigit():
return False
server.request_number = path
return None
def _options(server, path, i):
while len(path) != i and path[i].startswith('='):
server.options.append(path[i])
del path[i]
if len(path) == i or (len(path) == i+1 and path[i] == ''):
return ()
return False
SPECIALS = {
'{Y}': _year, '{ }': _time, '{S}': _semester, '{U}': _ue, '{P}': _page,
'{?}': _something, '{I}': _int, '{_I}': __int, '{*}': _anything,
'{=}': _options, '{R}': _request_number
}
plugins = [] # pylint: disable=invalid-name
class Plugin: # pylint: disable=too-many-instance-attributes
"""Define a plugin for TOMUSS server or 'suivi'"""
def __init__(self, name, url, function=lambda x: 0,
authenticated=True,
password_ok=True,
response=200,
mimetype="text/html; charset=UTF-8",
headers=lambda server: (),
launch_thread=False,
keep_open=False,
cached=False,
link=None,
documentation='',
css='',
priority=0,
group="",
unsafe=True,
# server.uploaded will contain the data
upload_max_size=0,
no_output=False, # The plugin writes an answer to browser
# Following parameters are deprecated, use group=groupname
teacher=None, referent=None, administrative=None,
abj_master=None, referent_master=None, root=None,
): # pylint: disable=too-many-arguments,too-many-locals,too-many-branches,too-many-statements
if url[0] != '/':
raise ValueError('not an absolute URL') # pragma: no cover
if '/' in name:
raise ValueError('/ is not allowed in plugin name') # pragma: no cover
for var in ('teacher', 'abj_master', 'referent_master', 'root',
'administrative', 'referent'):
value = locals()[var]
if value is not None: # pragma: no cover
if var == 'teacher':
var2 = 'staff'
else:
var2 = var + 's'
if value is False:
var2 = '!' + var2
utilities.warn("'%s=%s' is DEPRECATED, use 'group=%s'" %(var, value, var2),
what='Warning')
group = var2
self.name = name
self.url = url[1:].split('/')
if self.url == ['']:
self.url = []
self.function = function
self.authenticated = authenticated
if not authenticated:
unsafe = False
self.response = response
self.mimetype = mimetype
if mimetype and ("utf-8" in mimetype or "UTF-8" in mimetype):
self.codec = codecs.getwriter("utf-8")
else:
self.codec = lambda x: x
self.headers = headers
self.launch_thread = launch_thread
self.keep_open = keep_open
self.password_ok = password_ok
self.cached = cached
self.invited = ()
self.link = link
self.css = css
self.priority = priority
self.group = group
self.unsafe = unsafe
self.no_output = no_output
if upload_max_size and not launch_thread:
raise ValueError("Uploads must be in a thread") # pragma: no cover
if upload_max_size and 'html' not in mimetype:
raise ValueError("Uploads must return HTML mime type") # pragma: no cover
self.upload_max_size = upload_max_size
# Where the plugin is defined
self.module = sys._getframe(1).f_code.co_filename
if link:
link.set_plugin(self)
if documentation:
self.documentation = documentation
else:
self.documentation = function.__doc__
for plugin in plugins:
if plugin.name == self.name:
fct1 = plugin.function.__code__.co_filename
fct2 = self.function.__code__.co_filename
if fct1.split(os.path.sep)[-1] == fct2.split(os.path.sep)[-1]:
# __main__ module is loaded twice
continue
raise ValueError('Two plugins named "%s" (%s & %s)' %
(self.name, fct1, fct2)) # pragma: no cover
for plugin in plugins:
if plugin.name == self.name:
utilities.warn('Remove duplicate plugin name: %s' % plugin.name)
plugins.remove(plugin)
break
plugins.append(self)
plugins.sort(key=lambda x: (x.priority, x.url))
def __str__(self):
text = '%14s %-21s ' % (self.name, '/'.join(self.url))
text += {None: '', True:'Auth', False:'!Auth'}\
[self.authenticated].rjust(6)
text += {None: '', True:'PassOK', False:'!PassOK'}\
[self.password_ok].rjust(8)
text += {None: '', True:'LThrd', False:'!LThrd', 'batch':'batch'}\
[self.launch_thread].rjust(7)
text += ' ' + self.group
return text
def html(self):
"""Create the line for the developper documentation"""
text = '
'
if self.link:
text += '%s in %s %s' % (
html.escape(self.link.text),
self.link.where,
html.escape(self.link.help),
)
text += '
'
if self.documentation:
text += self.documentation
text += '
'
return text
def is_allowed(self, server):
"""Returns True if the plugin is allowed to the user"""
if not self.authenticated:
return True, 'Not authenticated'
if server.ticket is None:
return False, 'No ticket' # pragma: no cover
if self.password_ok is True and not server.ticket.password_ok:
return False, 'Only with good password'
if self.password_ok is False and server.ticket.password_ok:
return False, 'Only with bad password'
return (configuration.is_member_of(server.ticket.user_name,
self.invited),
'invited: ' + repr(self.invited))
def path_match(self, server):
"""Returns 'the_path' if the plugin is callable"""
path = server.the_path
url = self.url
path = list(path)
server.options = []
try:
for i, path_item in enumerate(url):
try:
the_path = SPECIALS[path_item](server, path, i)
if the_path is not None:
return the_path
except KeyError:
if path_item != path[i]:
return False
except (ValueError, IndexError):
return False
if len(path) == len(url):
return ()
return False
def backtrace_html(self):
"""To indicate the current plugin in the backtrace"""
return "Plugin: " + str(self)
def send_response(self, server):
"""Send the HTTP headers for this plugin"""
server.send_response(self.response)
if not self.cached:
server.send_header('Cache-Control', 'no-cache')
server.send_header('Cache-Control', 'no-store')
else:
server.send_header('Cache-Control',
'max-age=%d' % configuration.maxage)
for header in self.headers(server):
server.send_header(*header)
server.send_header('Content-Type', self.mimetype)
server.allow_cross_origin()
server.end_headers()
def vertical_text(text, size=12, exceptions=()):
"""Help function for the doc file: displays vertical text"""
if text in exceptions:
return text
height = str(int(size)*9)
size = str(size)
return '\n'
def create_html(filename):
"""Generate an HTML file with all the plugins for the developper"""
with open(filename, 'w', encoding="utf-8") as docfile:
docfile.write(
"
"
"
Name
"
"
URL template
"
"
" + vertical_text('Authenticated') + "
"
"
" + vertical_text('Group allowed') + "
"
"
" + vertical_text('Password OK') + "
"
"
" + vertical_text('Backgrounded') + "
"
"
" + vertical_text('Cached') + "
"
"
" + vertical_text('Keep open') + "
"
"
Mime Type
\n""")
for plugin in plugins:
docfile.write(plugin.html().replace('>N<', '> <') + "\n")
docfile.write("
\n")
def doc(filename):
"""Generate an HTML file with all the plugins for the administrator"""
uniq = {}
for plugin in plugins:
uniq[plugin.name] = plugin
uniq = list(uniq.values())
uniq.sort(key=lambda x: x.name)
with open(filename, 'w', encoding="utf-8") as docfile:
docfile.write('''
Name and location
Link and position
Explanations
''')
for plugin in uniq:
docfile.write(plugin.doc())
docfile.write('
')
# To add links on the home page, use the home configuration page
# on the home page.
LINKS_WITHOUT_PLUGINS = []
def add_links(*links):
"""Add the link if the url is not yet in the table"""
for new_link in links:
for link in LINKS_WITHOUT_PLUGINS:
if link.url == new_link.url:
break
else:
LINKS_WITHOUT_PLUGINS.append(new_link)
def get_links(server):
"""Get the home page links for the connected user"""
for link in LINKS_WITHOUT_PLUGINS:
if (configuration.is_member_of(server.ticket.user_name, link.group)
and link.authorized(server)):
if link.plugin and link.plugin() and not link.plugin().is_allowed(server)[0]:
continue # Not allowed by plugin
yield link, None
for plugin in plugins:
if (plugin.link
and plugin.is_allowed(server)[0]
and configuration.is_member_of(server.ticket.user_name,
plugin.link.group)
):
yield plugin.link, plugin
def bad_url(server): # pragma: no cover
"""The URL is bad"""
if configuration.regtest:
server.the_file.write('bad_url')
return
server.the_file.write(server._('ERROR_bad_url')
+ '%s' % (
configuration.server_url, server.ticket.ticket,
configuration.server_url
))
utilities.send_backtrace("PATH: %s\nREFERER: %s" % (
server.the_path,
server.headers.get("referer")),
"Bad URL", exception=False)
Plugin('bad-url', '/{url_not_possible}', function=bad_url, priority=1000, group='')
def execute(server, plugin): # pylint: disable=too-many-branches
"""Execute the plugin from a thread or HTTP server"""
if configuration.search_leak:
gc.collect(2) # pragma: no cover
if server.do_profile:
import cProfile # pylint: disable=import-outside-toplevel
profiler = cProfile.Profile(time.time)
profiler.enable()
if plugin.launch_thread:
server.the_file._sock.settimeout(60) # pylint: disable=protected-access
try:
important = None
try:
if plugin.mimetype and plugin.upload_max_size:
important = "uploading_%d" % id(server)
utilities.important_job_add(important)
configuration.time_of_last_upload = time.time()
server.uploaded = server.get_field_storage(
plugin.upload_max_size)
try:
plugin.send_response(server)
except AttributeError: # pragma: no cover
if not plugin.no_output:
raise
plugin.function(server)
finally:
if important:
utilities.important_job_remove(important)
except: # pylint: disable=bare-except
utilities.send_backtrace('Path = ' + str(server.the_path) + '\n' +
'Ticket = ' + str(server.ticket) + '\n',
subject='Plugin ' + plugin.name,
)
try:
if plugin.mimetype:
if 'image' in plugin.mimetype:
server.the_file.write(files.files['bug.png'].bytes()) # pragma: no cover
else:
server.the_file.write('*'*100 + " \n"
+ server._("ERROR_server_bug")
+ " \n" + '*'*100)
server.close_connection_now()
except: # pylint: disable=bare-except # pragma: no cover
pass
server.the_file = None
return
if not plugin.keep_open:
server.close_connection_now()
server.the_file = None
else:
plugin.function(server)
if server.do_profile:
profiler.disable()
profiler.dump_stats("xxx.prof")
import pstats # pylint: disable=import-outside-toplevel
stats = pstats.Stats('xxx.prof')
stats.strip_dirs().sort_stats('cumulative').print_stats()
if configuration.search_leak: # pragma: no cover
gc.set_debug(gc.DEBUG_SAVEALL)
nr_leaked = gc.collect(2)
gc.set_debug(0)
if nr_leaked:
utilities.warn("«{}» leak {} objects".format(plugin, nr_leaked))
for i in gc.garbage:
utilities.warn(str(i))
gc.garbage.clear()
server.log_time(plugin.name)
def search_plugin(server, manage_error):
"""Search an allowed plugin and create 'the_path'"""
for plugin in plugins:
if manage_error is False and plugin.authenticated:
# Only test plugin without authentication
continue
if plugin.is_allowed(server)[0]:
the_path = plugin.path_match(server)
if the_path is not False:
server.the_path = the_path
return plugin
return False
def get(name):
"""Get plugin from name"""
for plugin in plugins:
if plugin.name == name:
return plugin
return None
background_todo = []
def background_job():
"""Execute batch requests on after the other to not create many threads"""
done_time = time.time()
while background_todo:
fct, args = background_todo.pop(0)
fct(*args)
done_time = time.time()
return done_time
nr_plugin_call = 0
@utilities.add_a_lock
def dispatch_request(server, manage_error=True):
"""Search an executable plugin.
If manage_error=False, then we want to test if a plugin
not needing authentication is callable. Return False if not one is.
"""
global nr_plugin_call
nr_plugin_call += 1
unsafe = server.unsafe()
plugin = search_plugin(server, manage_error)
if plugin is False:
if manage_error:
plugin = get('bad-url')
else:
return False
if plugin.unsafe and unsafe: # pragma: no cover
server.send_response(plugin.response)
server.send_header('Content-Type', 'text/html; charset=UTF-8')
server.end_headers()
url = (authentication.authentication_redirect
+ server.path.replace("unsafe=1", "unsafe=0"))
to_send = server._('MSG_beware_XSS') + ' ' \
+ url.split("?")[0] + ''
server.the_file.write(to_send.encode("utf-8"))
server.the_file.close()
utilities.send_backtrace("XSS attack on " + server.ticket.user_name,
"URL: %s\nTICKET: %s" % (url, server.ticket),
exception=False)
return None
server.the_file = plugin.codec(server.the_file)
if plugin.mimetype and not plugin.upload_max_size:
plugin.send_response(server)
server.plugin = plugin
if plugin.keep_open or plugin.launch_thread:
server.do_not_close_connection()
if plugin.launch_thread:
if plugin.launch_thread is True:
utilities.start_new_thread(execute, (server, plugin))
else:
background_todo.append((execute, (server, plugin)))
utilities.start_job(background_job, 0)
else:
execute(server, plugin)
return None