moth/devel/devel-server.py

#!/usr/bin/python3

import asyncio
import cgitb
import html
import cgi
import http.server
import io
import json
import mimetypes
import moth
import logging
import os
import pathlib
import random
import shutil
import sys
import traceback
import mothballer
import parse
import urllib.parse
import posixpath
from http import HTTPStatus


sys.dont_write_bytecode = True  # Don't write .pyc files

try:
    ThreadingHTTPServer = http.server.ThreadingHTTPServer
except AttributeError:
    import socketserver
    class ThreadingHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
        daemon_threads = True

class MothServer(ThreadingHTTPServer):
    def __init__(self, server_address, RequestHandlerClass):
        super().__init__(server_address, RequestHandlerClass)
        self.args = {}


class MothRequestHandler(http.server.SimpleHTTPRequestHandler):
    endpoints = []
    
    def __init__(self, request, client_address, server):
        self.directory = str(server.args["theme_dir"])
        super().__init__(request, client_address, server)

    # Backport from Python 3.7
    def translate_path(self, path):
        """Translate a /-separated PATH to the local filename syntax.
        Components that mean special things to the local file system
        (e.g. drive or directory names) are ignored.  (XXX They should
        probably be diagnosed.)
        """
        # abandon query parameters
        path = path.split('?',1)[0]
        path = path.split('#',1)[0]
        # Don't forget explicit trailing slash when normalizing. Issue17324
        trailing_slash = path.rstrip().endswith('/')
        try:
            path = urllib.parse.unquote(path, errors='surrogatepass')
        except UnicodeDecodeError:
            path = urllib.parse.unquote(path)
        path = posixpath.normpath(path)
        words = path.split('/')
        words = filter(None, words)
        path = self.directory
        for word in words:
            if os.path.dirname(word) or word in (os.curdir, os.pardir):
                # Ignore components that are not a simple file/directory name
                continue
            path = os.path.join(path, word)
        if trailing_slash:
            path += '/'
        return path


    def get_puzzle(self):
        category = self.req.get("cat")
        points = int(self.req.get("points"))
        catpath = str(self.server.args["puzzles_dir"].joinpath(category))
        cat = moth.Category(catpath, self.seed)
        puzzle = cat.puzzle(points)
        return puzzle


    def handle_answer(self):
        for f in ("cat", "points", "answer"):
            self.req[f] = self.fields.getfirst(f)
        puzzle = self.get_puzzle()
        ret = {
            "status": "success",
            "data": {
               "short": "",
               "description": "Provided answer was not in list of answers"
            },
        }

        if self.req.get("answer") in puzzle.answers:
            ret["data"]["description"] = "Answer is correct"
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(ret).encode("utf-8"))
    endpoints.append(('/{seed}/answer', handle_answer))

    
    def handle_puzzlelist(self):
        puzzles = {
            "__devel__": [[0, ""]],
        }
        for p in self.server.args["puzzles_dir"].glob("*"):
            if not p.is_dir() or p.match(".*"):
                continue
            catName = p.parts[-1]
            cat = moth.Category(str(p), self.seed)
            puzzles[catName] = [[i, str(i)] for i in cat.pointvals()]
            puzzles[catName].append([0, ""])
        if len(puzzles) <= 1:
            logging.warning("No directories found matching {}/*".format(self.server.args["puzzles_dir"]))
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(puzzles).encode("utf-8"))
    endpoints.append(('/{seed}/puzzles.json', handle_puzzlelist))
    
    
    def handle_puzzle(self):
        puzzle = self.get_puzzle()

        obj = puzzle.package()
        obj["answers"] = puzzle.answers
        obj["hint"] = puzzle.hint
        obj["summary"] = puzzle.summary
        obj["logs"] = puzzle.logs

        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(obj).encode("utf-8"))
    endpoints.append(('/{seed}/content/{cat}/{points}/puzzle.json', handle_puzzle))


    def handle_puzzlefile(self):
        puzzle = self.get_puzzle()

        try:
            file = puzzle.files[self.req["filename"]]
        except KeyError:
            self.send_error(
                HTTPStatus.NOT_FOUND,
                "File Not Found",
            )
            return

        self.send_response(200)
        self.send_header("Content-Type", mimetypes.guess_type(file.name))
        self.end_headers()
        shutil.copyfileobj(file.stream, self.wfile)
    endpoints.append(("/{seed}/content/{cat}/{points}/{filename}", handle_puzzlefile))
    

    def handle_mothballer(self):
        category = self.req.get("cat")
        
        try:
            catdir = self.server.args["puzzles_dir"].joinpath(category)
            mb = mothballer.package(category, catdir, self.seed)
        except:
            self.send_response(200)
            self.send_header("Content-Type", "text/html; charset=\"utf-8\"")
            self.end_headers()
            self.wfile.write(cgitb.html(sys.exc_info()))
            return
        
        self.send_response(200)
        self.send_header("Content-Type", "application/octet_stream")
        self.end_headers()
        shutil.copyfileobj(mb, self.wfile)
    endpoints.append(("/{seed}/mothballer/{cat}.mb", handle_mothballer))


    def handle_index(self):
        seed = random.getrandbits(32)
        body = """<!DOCTYPE html>
<html>
  <head>
    <title>Dev Server</title>
    <script>
// Skip trying to log in
sessionStorage.setItem("id", "devel-server")
    </script>
  </head>
  <body>
    <h1>Dev Server</h1>

    <p>
      Pick a seed:
    </p>
    <ul>
      <li><a href="{seed}/">{seed}</a>: a special seed I made just for you!</li>
      <li><a href="random/">random</a>: will use a different seed every time you load a page (could be useful for debugging)</li>
      <li>You can also hack your own seed into the URL, if you want to.</li>
    </ul>

    <p>
      Puzzles can be generated from Python code: these puzzles can use a random number generator if they want.
      The seed is used to create these random numbers.
    </p>
    
    <p>
      We like to make a new seed for every contest,
      and re-use that seed whenever we regenerate a category during an event
      (say to fix a bug).
      By using the same seed,
      we make sure that all the dynamically-generated puzzles have the same values
      in any new packages we build.
    </p>
  </body>
</html>
""".format(seed=seed)

        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.end_headers()
        self.wfile.write(body.encode('utf-8'))
    endpoints.append((r"/", handle_index))


    def handle_theme_file(self):
        self.path = "/" + self.req.get("path", "")
        super().do_GET()
    endpoints.append(("/{seed}/", handle_theme_file))
    endpoints.append(("/{seed}/{path}", handle_theme_file))


    def do_GET(self):
        self.fields = cgi.FieldStorage(
            fp=self.rfile,
            headers=self.headers,
            environ={
                "REQUEST_METHOD": self.command,
                "CONTENT_TYPE": self.headers["Content-Type"],
            },
        )

        for pattern, function in self.endpoints:
            result = parse.parse(pattern, self.path)
            if result:
                self.req = result.named
                seed = self.req.get("seed", "random")
                if seed == "random":
                    self.seed = random.getrandbits(32)
                else:
                    self.seed = int(seed)
                return function(self)
        super().do_GET()

    def do_POST(self):
        self.do_GET()

    def do_HEAD(self):
        self.send_error(
            HTTPStatus.NOT_IMPLEMENTED,
            "Unsupported method (%r)" % self.command,
        )


if __name__ == '__main__':
    import argparse

    parser = argparse.ArgumentParser(description="MOTH puzzle development server")
    parser.add_argument(
        '--puzzles', default='puzzles',
        help="Directory containing your puzzles"
    )
    parser.add_argument(
        '--theme', default='theme',
        help="Directory containing theme files")
    parser.add_argument(
        '--bind', default="127.0.0.1:8080",
        help="Bind to ip:port"
    )
    parser.add_argument(
        '--base', default="",
        help="Base URL to this server, for reverse proxy setup"
    )
    args = parser.parse_args()
    parts = args.bind.split(":")
    addr = parts[0] or "0.0.0.0"
    port = int(parts[1])
    
    logging.basicConfig(level=logging.INFO)
    
    server = MothServer((addr, port), MothRequestHandler)
    server.args["base_url"] = args.base
    server.args["puzzles_dir"] = pathlib.Path(args.puzzles)
    server.args["theme_dir"] = args.theme

    logging.info("Listening on %s:%d", addr, port)
    server.serve_forever()