#!/usr/bin/env python
# -*- 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
import socket
import re
import cgi
import os
import sys
from . import configuration
from . import utilities
from . import files
warn = utilities.warn
class Link(object):
def __init__(self,
text=None,
help='',
url=None,
target="_blank",
html_class='leaves',
priority=0,
where = None,
group = "",
key = None, # See TEMPLATES/config_home.py
# DEPRECATED, use 'group'
authorized = None
):
self.text = text
self.url = url
self.target = target
self.help = re.sub('[ \t\n]+',' ',help)
self.html_class = html_class
self.priority = priority
self.where = where
self.plugin = None
self.group = group
self.key = key
if authorized:
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)
def get_url_and_target(self, plugin):
if self.target:
target = ' target="' + self.target + '"'
else:
target = ''
url = self.url
if url is None:
if len(plugin.url) != 1:
raise ValueError('BUG: "%s" %s' % (plugin, self))
url = '/' + plugin.url[0]
if url.startswith('javascript:'):
target = ''
elif url.startswith('/'):
target = ''
return url, target
def get_text(self):
if self.text:
return self.text
if self.plugin:
return utilities._('LINK_' + self.plugin.name)
else:
return utilities._('LINK_' + self.url)
# Helper functions for parsing URL
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(dummy_server, path, i): return True, 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):
try:
server.the_page = int(path[i])
except ValueError:
return False
def _int(server, path, i):
if not path[i][-1].isdigit():
return False
server.the_student = utilities.the_login(path[i])
def __int(server, path, i):
if not path[i].startswith('_'):
return False
server.the_student = utilities.the_login(path[i][1:])
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 True, ()
else:
return False
specials = {
'{Y}': _year, '{ }': _time, '{S}': _semester, '{U}': _ue, '{P}': _page,
'{?}': _something, '{I}': _int, '{_I}': __int, '{*}': _anything,
'{=}': _options
}
plugins = []
class Plugin(object):
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,
# Following parameters are deprecated, use group=groupname
teacher=None, referent=None, administrative=None,
abj_master=None, referent_master=None, root=None,
):
if url[0] != '/':
raise ValueError('not an absolute URL')
for var in ('teacher', 'abj_master', 'referent_master', 'root',
'administrative', 'referent'):
value = locals()[var]
if value is not None:
if var == 'teacher':
var2 = 'staff'
else:
var2 = var + 's'
if value is False:
var2 = '!' + var2
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
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
# Where the plugin is defined
self.module = sys._getframe(1).f_code.co_filename
if link:
link.plugin = self
if documentation:
self.documentation = documentation
else:
self.documentation = function.__doc__
for plugin in plugins:
if plugin.name == self.name:
f1 = plugin.function.func_code.co_filename
f2 = self.function.func_code.co_filename
if f1.split(os.path.sep)[-1] == f2.split(os.path.sep)[-1]:
# __main__ module is loaded twice
continue
raise ValueError('Two plugins named "%s" (%s & %s)' %
(self.name, f1, f2))
for p in plugins:
if p.name == self.name:
utilities.warn('Remove duplicate plugin name: %s' % p.name)
plugins.remove(p)
break
plugins.append(self)
plugins.sort(key=lambda x: (x.priority, x.url))
def __str__(self):
s = '%14s %-21s ' % (self.name, '/'.join(self.url))
s += {None: '', True:'Auth', False:'!Auth'}\
[self.authenticated].rjust(6)
s += {None: '', True:'PassOK', False:'!PassOK'}\
[self.password_ok].rjust(8)
s += {None: '', True:'LThrd', False:'!LThrd'}\
[self.launch_thread].rjust(7)
s += ' ' + self.group
return s
def html(self):
s = '
'
if self.link:
s += '%s in %s %s' % (
cgi.escape(self.link.get_text()),
self.link.where,
cgi.escape(self.link.help),
)
s += '
'
if self.documentation:
s += self.documentation
s += '
'
return s
def is_allowed(self, server):
if not self.authenticated:
return True, 'Not authenticated'
if server.ticket == None:
return False, 'No ticket'
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):
path = server.the_path
url = self.url
path = list(path)
server.options = []
try:
for i, f in enumerate(url):
try:
v = specials[f](server, path, i)
if v is not None:
return v
except KeyError:
if f != path[i]:
return False
except (ValueError, IndexError):
return False
if len(path) == len(url):
return True, ()
else:
return False
def backtrace_html(self):
return "Plugin: " + str(self)
def vertical_text(text, size=12):
height = str(int(size)*9)
size = str(size)
return '\n'
def html(filename):
f = open(filename, 'w')
f.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 p in plugins:
f.write(p.html().replace('>N<','> <')+"\n")
f.write("
\n")
f.close()
def doc(filename):
uniq = {}
for p in plugins:
uniq[p.name] = p
uniq = uniq.values()
uniq.sort(key=lambda x: x.name)
f = open(filename, 'w')
f.write('
')
f.write('
Name and location
Link and position
Explanations
')
for p in uniq:
f.write(p.doc())
f.write('
')
f.close()
# 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 link in links:
for t in links_without_plugins:
if t.url == link.url:
break
else:
links_without_plugins.append(link)
def remove_links(*links):
"""Remove the link from the table"""
for link in links:
for t in links_without_plugins:
if t.url == link.url:
links_without_plugins.remove(t)
break
def get_links(server):
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 not link.plugin.is_allowed(server)[0]:
continue # Not allowed by plugin
yield link, link.plugin
for p in plugins:
if (p.link
and p.is_allowed(server)[0]
and configuration.is_member_of(server.ticket.user_name,
p.link.group)
):
yield p.link, p
def bad_url(server):
"""The URL is bad"""
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)
def bad_url_message(server):
server.the_file.write('bad_url')
to_top = None
def execute(server, plugin):
if server.do_profile:
import cProfile
import time
pr = cProfile.Profile(time.time)
pr.enable()
if plugin.launch_thread:
try:
plugin.function(server)
except:
utilities.send_backtrace('Path = ' + str(server.the_path) + '\n' +
'Ticket = ' + str(server.ticket) + '\n',
subject = 'Plugin ' + plugin.name,
)
try:
if 'image' in plugin.mimetype:
server.the_file.write(files.files['bug.png'])
else:
server.the_file.write('*'*100 + " \n"
+ server._("ERROR_server_bug")
+ " \n" + '*'*100)
server.close_connection_now()
except:
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:
pr.disable()
pr.dump_stats("xxx.prof")
import pstats
ps = pstats.Stats('xxx.prof')
ps.strip_dirs().sort_stats('cumulative').print_stats()
server.log_time(plugin.name)
def search_plugin(server, manage_error):
for p in plugins:
if manage_error is False and p.authenticated:
# Only test plugin without authentication
continue
# warn('%s %s' % (p, p.is_allowed(server)))
if p.is_allowed(server)[0]:
t = p.path_match(server)
if t != False:
server.the_path = t[1]
return p
return False
def get(name):
"""Get plugin from name"""
for p in plugins:
if p.name == name:
return p
@utilities.add_a_lock
def dispatch_request(server, manage_error=True):
warn('dispatch %s' % server.the_path, what='debug')
unsafe = server.unsafe()
p = search_plugin(server, manage_error)
if p is False:
if manage_error:
global to_top
if to_top is None:
# Here because the regtest value is now fine
if configuration.regtest:
to_top = Plugin('bad-url', '/{url_not_possible}',
function = bad_url_message,
priority=1)
else:
to_top = Plugin('bad-url', '/{url_not_possible}',
function = bad_url,
priority=1)
to_top.invited = ('grp:',)
p = to_top
else:
return False
if p.unsafe and unsafe:
server.send_response(p.response)
server.send_header('Content-Type', 'text/html; charset=UTF-8')
server.end_headers()
from . import authentication
url = (authentication.authentication_redirect
+ server.path.replace("unsafe=1", "unsafe=0"))
server.the_file.write(server._('MSG_beware_XSS')
+ ' '
+ url.split("?")[0] + '')
server.the_file.close()
utilities.send_backtrace("XSS attack on " + server.ticket.user_name,
"URL: %s\nTICKET: %s" % (url, server.ticket),
exception=False)
return
try:
server.the_ue = server.the_ue
except AttributeError:
pass
warn('match %s(%s), path: %s' % (p.name, p.url, server.the_path),
what='plugin')
if p.mimetype:
server.send_response(p.response)
if not p.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 h in p.headers(server):
warn('send header: %s' % str(h), what='plugin')
server.send_header(*h)
server.send_header('Content-Type', p.mimetype)
server.end_headers()
server.plugin = p
if p.keep_open or p.launch_thread:
server.do_not_close_connection()
server.please_do_not_close = True
warn('keep_open (closed=%s)' % server.the_file.closed, what='plugin')
if p.launch_thread:
warn('launch thread with %s' % server.the_file, what='plugin')
utilities.start_new_thread(execute, (server, p))
else:
execute(server, p)
warn('done', what='plugin')