#!/usr/bin/env python3 # python standard library from os import path import sys import base64 import uuid import logging.config import argparse import mimetypes import signal import asyncio import functools # user installed libraries import tornado.ioloop import tornado.web import tornado.httpserver from tornado.escape import to_unicode # this project from learnapp import LearnApp from tools import load_yaml, md_to_html # ---------------------------------------------------------------------------- # Decorator used to restrict access to the administrator # ---------------------------------------------------------------------------- def admin_only(func): @functools.wraps(func) def wrapper(self, *args, **kwargs): if self.current_user != '0': raise tornado.web.HTTPError(403) # forbidden else: func(self, *args, **kwargs) return wrapper # ============================================================================ # WebApplication - Tornado Web Server # ============================================================================ class WebApplication(tornado.web.Application): def __init__(self, learnapp, debug=False): handlers = [ (r'/login', LoginHandler), (r'/logout', LogoutHandler), (r'/change_password', ChangePasswordHandler), (r'/question', QuestionHandler), # renders each question (r'/topic/(.+)', TopicHandler), # start a topic (r'/file/(.+)', FileHandler), # serve files, images, etc (r'/', RootHandler), # show list of topics ] settings = { 'template_path': path.join(path.dirname(__file__), 'templates'), 'static_path': path.join(path.dirname(__file__), 'static'), 'static_url_prefix': '/static/', 'xsrf_cookies': True, 'cookie_secret': base64.b64encode(uuid.uuid4().bytes), 'login_url': '/login', 'debug': debug, } super().__init__(handlers, **settings) self.learn = learnapp # ============================================================================ # Handlers # ============================================================================ # ---------------------------------------------------------------------------- # Base handler common to all handlers. # ---------------------------------------------------------------------------- class BaseHandler(tornado.web.RequestHandler): @property def learn(self): return self.application.learn def get_current_user(self): cookie = self.get_secure_cookie('user') if cookie: counter = self.get_secure_cookie('counter') uid = cookie.decode('utf-8') if counter.decode('utf-8') == str(self.learn.get_login_counter(uid)): return uid # ---------------------------------------------------------------------------- # /auth/login and /auth/logout # ---------------------------------------------------------------------------- class LoginHandler(BaseHandler): def get(self): self.render('login.html', error='') async def post(self): uid = self.get_body_argument('uid') pw = self.get_body_argument('pw') login_ok = await self.learn.login(uid, pw) if login_ok: self.set_secure_cookie('user', uid) # expires_days=30 self.set_secure_cookie('counter', str(self.learn.get_login_counter(uid))) self.redirect('/') else: self.render('login.html', error='Número ou senha incorrectos') # ---------------------------------------------------------------------------- class LogoutHandler(BaseHandler): @tornado.web.authenticated def get(self): self.clear_cookie('user') self.clear_cookie('counter') self.redirect('/') def on_finish(self): self.learn.logout(self.current_user) # ---------------------------------------------------------------------------- class ChangePasswordHandler(BaseHandler): @tornado.web.authenticated async def post(self): uid = self.current_user pw = self.get_body_arguments('new_password')[0] changed_ok = await self.learn.change_password(uid, pw) if changed_ok: notification = to_unicode(self.render_string('notification.html', type='success', msg='A password foi alterada!')) else: notification = to_unicode(self.render_string('notification.html', type='danger', msg='A password não foi alterada!')) self.write({'msg': notification}) # ---------------------------------------------------------------------------- # Main page: / # Shows a list of topics and proficiency (stars, locked). # ---------------------------------------------------------------------------- class RootHandler(BaseHandler): @tornado.web.authenticated def get(self): uid = self.current_user self.render('maintopics-table.html', uid=uid, name=self.learn.get_student_name(uid), state=self.learn.get_student_state(uid), title=self.learn.get_title(), ) # ---------------------------------------------------------------------------- # Start a given topic: /topic/... # ---------------------------------------------------------------------------- class TopicHandler(BaseHandler): SUPPORTED_METHODS = ['GET'] @tornado.web.authenticated async def get(self, topic): uid = self.current_user try: await self.learn.start_topic(uid, topic) except KeyError: self.redirect('/') else: self.render('topic.html', uid=uid, name=self.learn.get_student_name(uid), ) # ---------------------------------------------------------------------------- # Serves files from the /public subdir of the topics. # Based on https://bhch.github.io/posts/2017/12/serving-large-files-with-tornado-safely-without-blocking/ # ---------------------------------------------------------------------------- # FIXME error in many situations... images are not shown... # seems to happen when the browser sends two GET requests at the same time class FileHandler(BaseHandler): SUPPORTED_METHODS = ['GET'] @tornado.web.authenticated # async def get(self, filename): uid = self.current_user public_dir = self.learn.get_current_public_dir(uid) filepath = path.expanduser(path.join(public_dir, filename)) try: f = open(filepath, 'rb') except FileNotFoundError: logging.error(f'File not found: {filepath}') except PermissionError: logging.error(f'No permission: {filepath}') else: content_type = mimetypes.guess_type(filename) self.set_header("Content-Type", content_type[0]) # divide the file into chunks and write one chunk at a time, so # that the write does not block the ioloop for very long. with f: data = f.read() self.write(data) # await self.flush() self.flush() # ---------------------------------------------------------------------------- # respond to AJAX to get a JSON question # ---------------------------------------------------------------------------- class QuestionHandler(BaseHandler): SUPPORTED_METHODS = ['GET', 'POST'] templates = { 'checkbox': 'question-checkbox.html', 'radio': 'question-radio.html', 'text': 'question-text.html', 'text-regex': 'question-text.html', 'numeric-interval': 'question-text.html', 'textarea': 'question-textarea.html', # -- information panels -- 'information': 'question-information.html', 'info': 'question-information.html', 'success': 'question-success.html', } # --- get question to render @tornado.web.authenticated def get(self): logging.debug('QuestionHandler.get()') user = self.current_user q = self.learn.get_current_question(user) if q is not None: question_html = to_unicode(self.render_string(self.templates[q['type']], question=q, md=md_to_html)) response = { 'method': 'new_question', 'params': { 'type': q['type'], 'question': question_html, 'progress': self.learn.get_student_progress(user), 'tries': q['tries'], } } else: finished_topic_html = to_unicode(self.render_string('finished_topic.html')) response = { 'method': 'finished_topic', 'params': { 'question': finished_topic_html } } self.write(response) # --- post answer, returns what to do next: shake, new_question, finished @tornado.web.authenticated async def post(self): logging.debug('QuestionHandler.post()') user = self.current_user answer = self.get_body_arguments('answer') # list # answers are returned in a list. fix depending on question type qtype = self.learn.get_student_question_type(user) if qtype in ('success', 'information', 'info'): answer = None elif qtype == 'radio' and not answer: answer = None elif qtype != 'checkbox': # radio, text, textarea, ... answer = answer[0] # check answer (nonblocking) and get corrected question q, action = await self.learn.check_answer(user, answer) response = {'method': action, 'params': {}} if action == 'right': # get next question in the topic comments_html = self.render_string('comments-right.html', comments=q['comments'], md=md_to_html) response['params'] = { 'type': q['type'], 'progress': self.learn.get_student_progress(user), 'comments': to_unicode(comments_html), 'tries': q['tries'], } elif action == 'try_again': comments_html = self.render_string('comments.html', comments=q['comments'], md=md_to_html) response['params'] = { 'type': q['type'], 'progress': self.learn.get_student_progress(user), 'comments': to_unicode(comments_html), 'tries': q['tries'], } elif action == 'wrong': # no more tries comments_html = to_unicode(self.render_string('comments.html', comments=q['comments'], md=md_to_html)) solution_html = to_unicode(self.render_string('solution.html', solution=q['solution'], md=md_to_html)) response['params'] = { 'type': q['type'], 'progress': self.learn.get_student_progress(user), 'comments': comments_html, 'solution': solution_html, 'tries': q['tries'], } else: logging.error(f'Unknown action: {action}') self.write(response) # ---------------------------------------------------------------------------- # Signal handler to catch Ctrl-C and abort server # ---------------------------------------------------------------------------- def signal_handler(signal, frame): r = input(' --> Stop webserver? (yes/no) ').lower() if r == 'yes': tornado.ioloop.IOLoop.current().stop() logging.critical('Webserver stopped.') sys.exit(0) else: logging.info('Abort canceled...') # ------------------------------------------------------------------------- # Tornado web server # ---------------------------------------------------------------------------- def main(): # --- Commandline argument parsing argparser = argparse.ArgumentParser(description='Server for online learning. Enrolled students and topics have to be previously configured. Please read the documentation included with this software before running the server.') argparser.add_argument('conffile', type=str, nargs='+', help='Topics configuration file in YAML format.') argparser.add_argument('--prefix', type=str, default='demo', help='Path prefix under which the topic directories can be found, e.g. ~/topics') argparser.add_argument('--port', type=int, default=8443, help='Port to be used by the HTTPS server, e.g. 8443') argparser.add_argument('--db', type=str, default='students.db', help='SQLite3 database file, e.g. students.db') argparser.add_argument('--debug', action='store_true', help='Enable debug messages') arg = argparser.parse_args() # --- Setup logging logger_file = 'logger-debug.yaml' if arg.debug else 'logger.yaml' SERVER_PATH = path.dirname(path.realpath(__file__)) LOGGER_CONF = path.join(SERVER_PATH, f'config/{logger_file}') try: logging.config.dictConfig(load_yaml(LOGGER_CONF)) except: print('An error ocurred while setting up the logging system.') sys.exit(1) logging.info('====================================================') # --- start application logging.info('Starting App') try: learnapp = LearnApp(arg.conffile, prefix=arg.prefix, db=arg.db) except Exception as e: logging.critical('Failed to start backend application') raise e # --- create web application logging.info('Starting Web App (tornado)') try: webapp = WebApplication(learnapp, debug=arg.debug) except Exception as e: logging.critical('Failed to start web application.') raise e # --- create webserver try: http_server = tornado.httpserver.HTTPServer(webapp, ssl_options={ "certfile": "certs/cert.pem", "keyfile": "certs/privkey.pem" }) except ValueError: logging.critical('Certificates cert.pem and privkey.pem not found') sys.exit(1) else: http_server.listen(arg.port) logging.info(f'Listening on port {arg.port}.') # --- run webserver logging.info('Webserver running... (Ctrl-C to stop)') signal.signal(signal.SIGINT, signal_handler) try: tornado.ioloop.IOLoop.current().start() # running... except Exception: logging.critical('Webserver stopped.') tornado.ioloop.IOLoop.current().stop() raise # ---------------------------------------------------------------------------- if __name__ == "__main__": main()