#!/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 # 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), # page for exercising 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: user = cookie.decode('utf-8') return user # FIXME if the cookie exists but user is not in learn.online, this will force new login and store new (duplicate?) cookie. is this correct?? # if user in self.learn.online: # return user # ---------------------------------------------------------------------------- # /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", str(uid), expires_days=30) self.redirect(self.get_argument("next", "/")) 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.redirect(self.get_argument('next', '/')) 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 = tornado.escape.to_unicode(self.render_string('notification.html', type='success', msg='A password foi alterada!')) else: notification = tornado.escape.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(), # get_topic_type=self.learn.get_topic_type, # function ) # ---------------------------------------------------------------------------- # Start a given topic: /topic/... # ---------------------------------------------------------------------------- class TopicHandler(BaseHandler): @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/ # ---------------------------------------------------------------------------- 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: self.write(f.read()) await self.flush() # ---------------------------------------------------------------------------- # respond to AJAX to get a JSON question # ---------------------------------------------------------------------------- class QuestionHandler(BaseHandler): 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 question = self.learn.get_current_question(user) question_html = self.render_string(self.templates[question['type']], question=question, md=md_to_html) self.write({ 'method': 'new_question', 'params': { 'question': tornado.escape.to_unicode(question_html), 'progress': self.learn.get_student_progress(user), 'tries': question['tries'], }, }) # --- 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 in another thread (nonblocking) action = await self.learn.check_answer(user, answer) # get next question (same, new or None) question = self.learn.get_current_question(user) if action == 'wrong': comments_html = self.render_string('comments.html', comments=question['comments'], md=md_to_html) self.write({ 'method': action, 'params': { 'progress': self.learn.get_student_progress(user), 'comments': tornado.escape.to_unicode(comments_html), # FIXME 'tries': question['tries'], } }) elif action == 'finished_topic': # right answer, finished topic finished_topic_html = self.render_string('finished_topic.html') self.write({ 'method': 'finished_topic', 'params': { 'question': tornado.escape.to_unicode(finished_topic_html) } }) elif action == 'new_question': # get next question in the topic template = self.templates[question['type']] question_html = self.render_string(template, question=question, md=md_to_html) self.write({ 'method': 'new_question', 'params': { 'question': tornado.escape.to_unicode(question_html), 'progress': self.learn.get_student_progress(user), 'tries': question['tries'], } }) else: logger.error(f'Unknown action {action}') # ---------------------------------------------------------------------------- # 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('--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) 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()