From 8e8f1591506dacd398ac1cf07d06d91dedbab8d0 Mon Sep 17 00:00:00 2001 From: s3lph Date: Fri, 22 Nov 2019 03:58:54 +0100 Subject: [PATCH] Migrate from own web framework to bottle.py --- matemat/__main__.py | 72 +- matemat/webserver/__init__.py | 5 +- matemat/webserver/config.py | 49 +- matemat/webserver/cron.py | 85 +++ matemat/webserver/httpd.py | 636 ------------------ matemat/webserver/logger.py | 25 + matemat/webserver/pagelets/__init__.py | 1 - matemat/webserver/pagelets/admin.py | 81 ++- matemat/webserver/pagelets/buy.py | 30 +- matemat/webserver/pagelets/deposit.py | 30 +- matemat/webserver/pagelets/initialization.py | 53 -- matemat/webserver/pagelets/login.py | 46 +- matemat/webserver/pagelets/logout.py | 24 +- matemat/webserver/pagelets/main.py | 35 +- matemat/webserver/pagelets/modproduct.py | 67 +- matemat/webserver/pagelets/moduser.py | 67 +- .../webserver/pagelets/receipt_smtp_cron.py | 31 +- matemat/webserver/pagelets/statistics.py | 57 +- matemat/webserver/pagelets/touchkey.py | 48 +- matemat/webserver/pagelets/userbootstrap.py | 41 +- matemat/webserver/requestargs.py | 324 --------- matemat/webserver/responses.py | 65 -- matemat/webserver/session/__init__.py | 2 + matemat/webserver/session/sessions.py | 73 ++ matemat/webserver/template/__init__.py | 2 + matemat/webserver/template/template.py | 24 + matemat/webserver/test/test_pagelet_cron.py | 6 +- matemat/webserver/test/test_parse_request.py | 437 ------------ matemat/webserver/test/test_post.py | 225 ------- matemat/webserver/test/test_requestargs.py | 529 --------------- matemat/webserver/test/test_serve.py | 332 --------- matemat/webserver/util.py | 138 ---- setup.py | 1 + templates/admin_all.html | 2 +- templates/base.html | 2 +- templates/touchkey.html | 2 +- 36 files changed, 580 insertions(+), 3067 deletions(-) create mode 100644 matemat/webserver/cron.py delete mode 100644 matemat/webserver/httpd.py create mode 100644 matemat/webserver/logger.py delete mode 100644 matemat/webserver/pagelets/initialization.py delete mode 100644 matemat/webserver/requestargs.py delete mode 100644 matemat/webserver/responses.py create mode 100644 matemat/webserver/session/__init__.py create mode 100644 matemat/webserver/session/sessions.py create mode 100644 matemat/webserver/template/__init__.py create mode 100644 matemat/webserver/template/template.py delete mode 100644 matemat/webserver/test/test_parse_request.py delete mode 100644 matemat/webserver/test/test_post.py delete mode 100644 matemat/webserver/test/test_requestargs.py delete mode 100644 matemat/webserver/test/test_serve.py delete mode 100644 matemat/webserver/util.py diff --git a/matemat/__main__.py b/matemat/__main__.py index b10f837..8a773f7 100644 --- a/matemat/__main__.py +++ b/matemat/__main__.py @@ -1,27 +1,87 @@ from typing import Any, Dict, Iterable, Union +from bottle import route, static_file, run import sys +import os.path -from matemat.webserver import MatematWebserver -from matemat.webserver import parse_config_file +from matemat.db import MatematDatabase +from matemat.webserver.logger import Logger +from matemat.webserver.config import get_config, parse_config_file +from matemat.webserver.template import init as template_init # Those imports are actually needed, as they implicitly register pagelets. # noinspection PyUnresolvedReferences from matemat.webserver.pagelets import * +def _init(config: Dict[str, Any]): + logger = Logger.instance() + # Set default values for missing config items + if 'InstanceName' not in config: + config['InstanceName'] = 'Matemat' + logger.warning('Property \'InstanceName\' not set, using \'Matemat\'') + if 'UploadDir' not in config: + config['UploadDir'] = './static/upload/' + logger.warning('Property \'UploadDir\' not set, using \'./static/upload/\'') + if 'DatabaseFile' not in config: + config['DatabaseFile'] = './matemat.db' + logger.warning('Property \'DatabaseFile\' not set, using \'./matemat.db\'') + if 'SmtpSendReceipts' not in config: + config['SmtpSendReceipts'] = '0' + logger.warning('Property \'SmtpSendReceipts\' not set, using \'0\'') + if config['SmtpSendReceipts'] == '1': + if 'SmtpFrom' not in config: + logger.fatal('\'SmtpSendReceipts\' set to \'1\', but \'SmtpFrom\' missing.') + raise KeyError() + if 'SmtpSubj' not in config: + logger.fatal('\'SmtpSendReceipts\' set to \'1\', but \'SmtpSubj\' missing.') + raise KeyError() + if 'SmtpHost' not in config: + logger.fatal('\'SmtpSendReceipts\' set to \'1\', but \'SmtpHost\' missing.') + raise KeyError() + if 'SmtpPort' not in config: + logger.fatal('\'SmtpSendReceipts\' set to \'1\', but \'SmtpPort\' missing.') + raise KeyError() + if 'SmtpUser' not in config: + logger.fatal('\'SmtpSendReceipts\' set to \'1\', but \'SmtpUser\' missing.') + raise KeyError() + if 'SmtpPass' not in config: + logger.fatal('\'SmtpSendReceipts\' set to \'1\', but \'SmtpPass\' missing.') + raise KeyError() + if 'SmtpEnforceTLS' not in config: + config['SmtpEnforceTLS'] = '1' + logger.warning('Property \'SmtpEnforceTLS\' not set, using \'1\'') + with MatematDatabase(config['DatabaseFile']): + # Connect to the database to create it and perform any schema migrations + pass + # Initialize Jinaj2 template system + template_init(config) + + +@route('/static/') +def serve_static_files(filename: str): + config = get_config() + staticroot = os.path.abspath(config['staticroot']) + return static_file(filename, root=staticroot) + + def main(): - # Use config file name from command line, if present + # Use config file name from command line, if present, and parse it configfile: Union[str, Iterable[str]] = '/etc/matemat.conf' if len(sys.argv) > 1: configfile = sys.argv[1:] + parse_config_file(configfile) - # Parse the config file - config: Dict[str, Any] = parse_config_file(configfile) + config = get_config() + + _init(config) + + host: str = config['listen'] + port: int = int(str(config['port'])) # Start the web server - MatematWebserver(**config).start() + run(host=host, port=port, debug=True) if __name__ == '__main__': diff --git a/matemat/webserver/__init__.py b/matemat/webserver/__init__.py index 4519023..1317495 100644 --- a/matemat/webserver/__init__.py +++ b/matemat/webserver/__init__.py @@ -6,7 +6,4 @@ API that can be used by 'pagelets' - single pages of a web service. If a reques server will attempt to serve the request with a static resource in a previously configured webroot directory. """ -from .requestargs import RequestArgument, RequestArguments -from .responses import PageletResponse, RedirectResponse, TemplateResponse -from .httpd import MatematWebserver, HttpHandler, pagelet, pagelet_init, pagelet_cron -from .config import parse_config_file +from .logger import Logger diff --git a/matemat/webserver/config.py b/matemat/webserver/config.py index 5ac6059..fe48322 100644 --- a/matemat/webserver/config.py +++ b/matemat/webserver/config.py @@ -7,6 +7,9 @@ import logging from configparser import ConfigParser +config: Dict[str, Any] = dict() + + def parse_logging(symbolic_level: str, symbolic_target: str) -> Tuple[int, logging.Handler]: """ Parse the LogLevel and LogTarget from configuration. @@ -45,32 +48,32 @@ def parse_logging(symbolic_level: str, symbolic_target: str) -> Tuple[int, loggi return level, target -def parse_config_file(paths: Union[str, Iterable[str]]) -> Dict[str, Any]: +def parse_config_file(paths: Union[str, Iterable[str]]) -> None: """ Parse the configuration file at the given path. :param paths: The config file(s) to parse. :return: A dictionary containing the parsed configuration. """ + global config # Set up default values - config: Dict[str, Any] = { - # Address to listen on - 'listen': '::', - # TCP port to listen on - 'port': 80, - # Root directory of statically served content - 'staticroot': '/var/matemat/static', - # Root directory of Jinja2 templates - 'templateroot': '/var/matemat/templates', - # Log level - 'log_level': logging.INFO, - # Log target: An IO stream (stderr, stdout, ...) or a filename - 'log_handler': logging.StreamHandler(), - # Variables passed to pagelets - 'pagelet_variables': dict(), - # Statically configured headers - 'headers': dict() - } + + # Address to listen on + config['listen'] = '::' + # TCP port to listen on + config['port'] = 80 + # Root directory of statically served content + config['staticroot'] = '/var/matemat/static' + # Root directory of Jinja2 templates + config['templateroot'] = '/var/matemat/templates' + # Log level + config['log_level'] = logging.INFO + # Log target: An IO stream (stderr, stdout, ...) or a filename + config['log_handler'] = logging.StreamHandler() + # Variables passed to pagelets + config['pagelet_variables'] = dict() + # Statically configured headers + config['headers'] = dict() # Initialize the config parser parser: ConfigParser = ConfigParser() @@ -108,4 +111,12 @@ def parse_config_file(paths: Union[str, Iterable[str]]) -> Dict[str, Any]: for k, v in parser['HttpHeaders'].items(): config['headers'][k] = v + +def get_config() -> Dict[str, Any]: + global config return config + + +def get_app_config() -> Dict[str, Any]: + global config + return config['pagelet_variables'] diff --git a/matemat/webserver/cron.py b/matemat/webserver/cron.py new file mode 100644 index 0000000..3324d8a --- /dev/null +++ b/matemat/webserver/cron.py @@ -0,0 +1,85 @@ +from typing import Callable + +from datetime import timedelta +from threading import Event, Timer, Thread + +_CRON_STATIC_EVENT: Event = Event() + + +class _GlobalEventTimer(Thread): + """ + A timer similar to threading.Timer, except that waits on an externally supplied threading.Event instance, + therefore allowing all timers waiting on the same event to be cancelled at once. + """ + + def __init__(self, interval: float, event: Event, fun, *args, **kwargs): + """ + Create a new _GlobalEventTimer. + :param interval: The delay after which to run the function. + :param event: The external threading.Event to wait on. + :param fun: The function to call. + :param args: The positional arguments to pass to the function. + :param kwargs: The keyword arguments to pass to the function. + """ + Thread.__init__(self) + self.interval = interval + self.fun = fun + self.args = args if args is not None else [] + self.kwargs = kwargs if kwargs is not None else {} + self.event = event + + def run(self): + self.event.wait(self.interval) + if not self.event.is_set(): + self.fun(*self.args, **self.kwargs) + # Do NOT call event.set(), as done in threading.Timer, as that would cancel all other timers + + +def cron(weeks: int = 0, + days: int = 0, + hours: int = 0, + seconds: int = 0, + minutes: int = 0, + milliseconds: int = 0, + microseconds: int = 0): + """ + Annotate a function to act as a cron function. The function will be called in a regular interval, defined + by the arguments passed to the decorator, which are passed to a timedelta object. + + :param weeks: Number of weeks in the interval. + :param days: Number of days in the interval. + :param hours: Number of hours in the interval. + :param seconds: Number of seconds in the interval. + :param minutes: Number of minutes in the interval. + :param milliseconds: Number of milliseconds in the interval. + :param microseconds: Number of microseconds in the interval. + """ + + def cron_wrapper(fun: Callable[[], None]): + # Create the timedelta object + delta: timedelta = timedelta(weeks=weeks, + days=days, + hours=hours, + seconds=seconds, + minutes=minutes, + milliseconds=milliseconds, + microseconds=microseconds) + + # This function is called once in the specified interval + def cron(): + # Reschedule the job + t: Timer = _GlobalEventTimer(delta.total_seconds(), _CRON_STATIC_EVENT, cron) + t.start() + # Actually run the task + logger.info('Executing cron job "%s"', fun.__name__) + try: + fun() + logger.info('Completed cron job "%s"', fun.__name__) + except BaseException as e: + logger.exception('Cron job "%s" failed:', fun.__name__, exc_info=e) + + # Set a timer to run the cron job after the specified interval + timer: Timer = _GlobalEventTimer(delta.total_seconds(), _CRON_STATIC_EVENT, cron) + timer.start() + + return cron_wrapper diff --git a/matemat/webserver/httpd.py b/matemat/webserver/httpd.py deleted file mode 100644 index 8d72d61..0000000 --- a/matemat/webserver/httpd.py +++ /dev/null @@ -1,636 +0,0 @@ - -from typing import Any, Callable, Dict, Set, Tuple, Type, Union - -import logging -import os -import socket -import mimetypes -import magic -from socketserver import TCPServer -from http.server import HTTPServer, BaseHTTPRequestHandler -from http.cookies import SimpleCookie -from uuid import uuid4 -from datetime import datetime, timedelta -from threading import Event, Timer, Thread - -import jinja2 - -from matemat import __version__ as matemat_version -from matemat.exceptions import HttpException -from matemat.webserver import RequestArguments, PageletResponse, RedirectResponse, TemplateResponse -from matemat.webserver.util import parse_args -from matemat.util.currency_format import format_chf - -# -# Python internal class hacks -# - -# Enable IPv6 support (IPv6/IPv4 dual-stack support should be implicitly enabled) -TCPServer.address_family = socket.AF_INET6 - -# Dictionary to hold registered pagelet paths and their handler functions -_PAGELET_PATHS: Dict[str, Callable[[str, # HTTP method (GET, POST, ...) - str, # Request path - RequestArguments, # HTTP Request arguments - Dict[str, Any], # Session vars - Dict[str, str], # Response headers - Dict[str, str]], # Items from the [Pagelets] section in the config file - Union[ # Return type: either a response body, or a redirect - bytes, str, # Response body: will assign HTTP/1.0 200 OK - PageletResponse, # A generic response - ]]] = dict() -# The pagelet initialization functions, to be executed upon startup -_PAGELET_INIT_FUNCTIONS: Set[Callable[[Dict[str, str], logging.Logger], None]] = set() - -_PAGELET_CRON_STATIC_EVENT: Event = Event() -_PAGELET_CRON_RUNNER: Callable[[Callable[[Dict[str, str], jinja2.Environment, logging.Logger], None]], None] = None - -# Inactivity timeout for client sessions -_SESSION_TIMEOUT: int = 3600 -_MAX_POST: int = 1_000_000 - - -def pagelet(path: str): - """ - Annotate a function to act as a pagelet (part of a website). The function will be called if a request is made to - the path specified as argument to the annotation. - - The function must have the following signature: - - (method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - config: Dict[str, str]) - -> Union[bytes, str, Tuple[int, str]] - - method: The HTTP method (GET, POST) that was used. - path: The path that was requested. - args: The arguments that were passed with the request (as GET or POST arguments). - session_vars: The session storage. May be read from and written to. - headers: The dictionary of HTTP response headers. Add headers you wish to send with the response. - config: The dictionary of variables read from the [Pagelets] section of the configuration file. - returns: One of the following: - - A HTTP Response body as str or bytes - - A PageletResponse class instance: An instance of (a subclass of) - matemat.webserver.PageletResponse, e.g. encapsulating a redirect or a Jinja2 template. - raises: HttpException: If a non-200 HTTP status code should be returned - - :param path: The path to register the function for. - """ - - def http_handler(fun: Callable[[str, - str, - RequestArguments, - Dict[str, Any], - Dict[str, str], - Dict[str, str]], - Union[ - bytes, str, - PageletResponse - ]]): - # Add the function to the dict of pagelets - _PAGELET_PATHS[path] = fun - # Don't change the function itself at all - return fun - - # Return the inner function (Python requires a "real" function annotation to not have any arguments except - # the function itself) - return http_handler - - -def pagelet_init(fun: Callable[[Dict[str, str], logging.Logger], None]): - """ - Annotate a function to act as a pagelet initialization function. The function will be called when the webserver is - started. The function should set up everything required for pagelets to operate correctly, e.g. providing default - values for configuration items, or performing database schema migrations. - - Any exception thrown by an initialization function will cause the web server to seize operation. Multiple - initialization functions can be used. - - The function must have the following signature: - - (config: Dict[str, str], logger: logging.Logger) -> None - - config: The mutable dictionary of variables read from the [Pagelets] section of the configuration file. - logger: The server's logger instance. - returns: Nothing. - - :param fun: The function to annotate - """ - _PAGELET_INIT_FUNCTIONS.add(fun) - - -class _GlobalEventTimer(Thread): - """ - A timer similar to threading.Timer, except that waits on an externally supplied threading.Event instance, - therefore allowing all timers waiting on the same event to be cancelled at once. - """ - - def __init__(self, interval: float, event: Event, fun, *args, **kwargs): - """ - Create a new _GlobalEventTimer. - :param interval: The delay after which to run the function. - :param event: The external threading.Event to wait on. - :param fun: The function to call. - :param args: The positional arguments to pass to the function. - :param kwargs: The keyword arguments to pass to the function. - """ - Thread.__init__(self) - self.interval = interval - self.fun = fun - self.args = args if args is not None else [] - self.kwargs = kwargs if kwargs is not None else {} - self.event = event - - def run(self): - self.event.wait(self.interval) - if not self.event.is_set(): - self.fun(*self.args, **self.kwargs) - # Do NOT call event.set(), as done in threading.Timer, as that would cancel all other timers - - -def pagelet_cron(weeks: int = 0, - days: int = 0, - hours: int = 0, - seconds: int = 0, - minutes: int = 0, - milliseconds: int = 0, - microseconds: int = 0): - """ - Annotate a function to act as a pagelet cron function. The function will be called in a regular interval, defined - by the arguments passed to the decorator, which are passed to a timedelta object. - - The function must have the following signature: - - (config: Dict[str, str], jinja_env: jinja2.Environment, logger: logging.Logger) -> None - - config: The mutable dictionary of variables read from the [Pagelets] section of the configuration file. - jinja_env: The Jinja2 environment used by the web server. - logger: The server's logger instance. - returns: Nothing. - - :param weeks: Number of weeks in the interval. - :param days: Number of days in the interval. - :param hours: Number of hours in the interval. - :param seconds: Number of seconds in the interval. - :param minutes: Number of minutes in the interval. - :param milliseconds: Number of milliseconds in the interval. - :param microseconds: Number of microseconds in the interval. - """ - - def cron_wrapper(fun: Callable[[Dict[str, str], jinja2.Environment, logging.Logger], None]): - # Create the timedelta object - delta: timedelta = timedelta(weeks=weeks, - days=days, - hours=hours, - seconds=seconds, - minutes=minutes, - milliseconds=milliseconds, - microseconds=microseconds) - - # This function is called once in the specified interval - def cron(): - # Set a new timer - t: Timer = _GlobalEventTimer(delta.total_seconds(), _PAGELET_CRON_STATIC_EVENT, cron) - t.start() - # Have the cron job be picked up by the cron runner provided by the web server - if _PAGELET_CRON_RUNNER is not None: - _PAGELET_CRON_RUNNER(fun) - - # Set a timer to run the cron job after the specified interval - timer: Timer = _GlobalEventTimer(delta.total_seconds(), _PAGELET_CRON_STATIC_EVENT, cron) - timer.start() - - return cron_wrapper - - -class MatematHTTPServer(HTTPServer): - """ - A http.server.HTTPServer subclass that acts as a container for data that must be persistent between requests. - """ - - def __init__(self, - server_address: Any, - handler: Type[BaseHTTPRequestHandler], - staticroot: str, - templateroot: str, - pagelet_variables: Dict[str, str], - static_headers: Dict[str, str], - log_level: int, - log_handler: logging.Handler, - bind_and_activate: bool = True) -> None: - super().__init__(server_address, handler, bind_and_activate) - # Resolve webroot directory - self.webroot = os.path.abspath(staticroot) - # Set up session vars dict - self.session_vars: Dict[str, Tuple[datetime, Dict[str, Any]]] = dict() - # Set up pagelet arguments dict - self.pagelet_variables = pagelet_variables - # Set up static HTTP Response headers - self.static_headers = static_headers - # Set up the Jinja2 environment - self.jinja_env: jinja2.Environment = jinja2.Environment( - loader=jinja2.FileSystemLoader(os.path.abspath(templateroot)), - autoescape=jinja2.select_autoescape(default=True), - ) - self.jinja_env.filters['chf'] = format_chf - # Set up logger - self.logger: logging.Logger = logging.getLogger('matemat.webserver') - self.logger.setLevel(log_level) - # Set up the log handler's (obtained from config parsing) format string - log_handler.setFormatter(logging.Formatter('%(asctime)s %(name)s [%(levelname)s]: %(message)s')) - self.logger.addHandler(log_handler) - - -class MatematWebserver(object): - """ - Then main webserver class, internally uses Python's http.server. - - The server will serve a pagelet, if one is defined for a request path, else it will attempt to serve a static - resource from the webroot. - - Usage: - - # Listen on all interfaces on port 80 (dual-stack IPv6/IPv4) - server = MatematWebserver('::', 80, webroot='/var/www/matemat') - # Start the server. This call blocks while the server is running. - server.start() - """ - - def __init__(self, - listen: str, - port: int, - staticroot: str, - templateroot: str, - pagelet_variables: Dict[str, str], - headers: Dict[str, str], - log_level: int, - log_handler: logging.Handler) -> None: - """ - Instantiate a MatematWebserver. - - :param listen: The IPv4 or IPv6 address to listen on. - :param port: The TCP port to listen on. - :param staticroot: Path to the static webroot directory. - :param templateroot: Path to the Jinja2 templates root directory. - :param pagelet_variables: Dictionary of variables to pass to pagelet functions. - :param headers: Dictionary of statically configured headers. - :param log_level: The log level, as defined in the builtin logging module. - :param log_handler: The logging handler. - """ - # IPv4 address detection heuristic - if ':' not in listen and '.' in listen: - # Rewrite IPv4 address to IPv6-mapped form - listen = f'::ffff:{listen}' - # Create the http server - self._httpd = MatematHTTPServer((listen, port), - HttpHandler, - staticroot, - templateroot, - pagelet_variables, - headers, - log_level, - log_handler) - - def start(self) -> None: - """ - Call all pagelet initialization functions and start the web server. This call blocks while the server is - running. If any exception is raised in the initialization phase, the program is terminated with a non-zero - exit code. - """ - global _PAGELET_CRON_RUNNER - try: - try: - # Run all pagelet initialization functions - for fun in _PAGELET_INIT_FUNCTIONS: - fun(self._httpd.pagelet_variables, self._httpd.logger) - # Set pagelet cron runner to self - _PAGELET_CRON_RUNNER = self._cron_runner - except BaseException as e: - # If an error occurs, log it and terminate - self._httpd.logger.exception(e) - self._httpd.logger.critical('An initialization pagelet raised an error. Stopping.') - raise e - # If pagelet initialization went fine, start the HTTP server - self._httpd.serve_forever() - finally: - # Cancel all cron timers at once when the webserver is shutting down - _PAGELET_CRON_STATIC_EVENT.set() - - def _cron_runner(self, fun: Callable[[Dict[str, str], jinja2.Environment, logging.Logger], None]): - self._httpd.logger.info('Executing cron job "%s"', fun.__name__) - try: - fun(self._httpd.pagelet_variables, - self._httpd.jinja_env, - self._httpd.logger) - self._httpd.logger.info('Completed cron job "%s"', fun.__name__) - except BaseException as e: - self._httpd.logger.exception('Cron job "%s" failed:', fun.__name__, exc_info=e) - - -class HttpHandler(BaseHTTPRequestHandler): - """ - HTTP Request handler. - - This class parses HTTP requests, and calls the appropriate pagelets, or fetches a static resource from the webroot - directory. - """ - - def __init__(self, request: bytes, client_address: Tuple[str, int], server: HTTPServer) -> None: - super().__init__(request, client_address, server) - self.server: MatematHTTPServer - - @property - def server_version(self) -> str: - return f'matemat/{matemat_version}' - - def _start_session(self) -> Tuple[str, datetime]: - """ - Start a new session, or resume the session identified by the session cookie sent in the HTTP request. - - :return: A tuple consisting of the session ID (a UUID string), and the session timeout date. - """ - # Reference date for session timeout - now = datetime.utcnow() - # Parse cookies sent by the client - cookiestring = '\n'.join(self.headers.get_all('Cookie', failobj=[])) - cookie = SimpleCookie() - cookie.load(cookiestring) - # Read the client's session ID, if any - session_id = str(cookie['matemat_session_id'].value) if 'matemat_session_id' in cookie else None - # If there is no active session, create a new session ID - if session_id is None or session_id not in self.server.session_vars: - session_id = str(uuid4()) - self.server.logger.debug('Started session %s', session_id) - - # Check for session timeout - if session_id in self.server.session_vars and self.server.session_vars[session_id][0] < now: - self._end_session(session_id) - self.server.logger.debug('Session %s timed out', session_id) - raise TimeoutError('Session timed out.') - # Update or initialize the session timeout - if session_id not in self.server.session_vars: - self.server.session_vars[session_id] = (now + timedelta(seconds=_SESSION_TIMEOUT)), dict() - else: - self.server.session_vars[session_id] = \ - (now + timedelta(seconds=_SESSION_TIMEOUT), self.server.session_vars[session_id][1]) - # Return the session ID and timeout - return session_id, self.server.session_vars[session_id][0] - - def _end_session(self, session_id: str) -> None: - """ - Destroy a session identified by the session ID. - - :param session_id: ID of the session to destroy. - """ - if session_id in self.server.session_vars: - del self.server.session_vars[session_id] - - def _parse_pagelet_result(self, - pagelet_res: Union[bytes, # Response body as bytes - str, # Response body as str - PageletResponse], # Encapsulated or unresolved response body - headers: Dict[str, str]) \ - -> Tuple[int, bytes]: - """ - Process the return value of a pagelet function call. - - :param pagelet_res: The pagelet return value. - :param headers: The dict of HTTP response headers, needed for setting the redirect header. - :return: The HTTP Response status code (an int) and body (a bytes). - :raises TypeError: If the pagelet result was not in the expected form. - """ - # The HTTP Response Status Code, defaults to 200 OK - hsc: int = 200 - # The HTTP Response body, defaults to empty - data: bytes = bytes() - - # If the response is a bytes object, it is used without further modification - if isinstance(pagelet_res, bytes): - data = pagelet_res - - # If the response is a str object, it is encoded into a bytes object - elif isinstance(pagelet_res, str): - data = pagelet_res.encode('utf-8') - - # If the response is a PageletResponse object, the status code is extracted. Generation of the body depends - # on the subtype - elif isinstance(pagelet_res, PageletResponse): - hsc = pagelet_res.status - - # If the object is a RedirectResponse instance, no body is needed - if isinstance(pagelet_res, RedirectResponse): - headers['Location'] = pagelet_res.location - # If the object is a TemplateRespinse instance, pass the Jinja2 environment instance for rendering - elif isinstance(pagelet_res, TemplateResponse): - # noinspection PyProtectedMember - data = pagelet_res._render(self.server.jinja_env) - # else: Empty body - - else: - raise TypeError(f'Return value of pagelet not understood: {pagelet_res}') - - # Return the resulting status code and body - return hsc, data - - def _handle(self, method: str, path: str, args: RequestArguments) -> None: - """ - Handle a HTTP request by either dispatching it to the appropriate pagelet or by serving a static resource. - - :param method: The HTTP request method (GET, POST). - :param path: The request path without GET arguments. - :param args: Arguments sent with the request. This includes GET and POST arguments, where the POST arguments - take precedence. - """ - # Start or resume a session; redirect to / on session timeout - try: - session_id, timeout = self._start_session() - except TimeoutError: - self.send_response(302) - self.send_header('Set-Cookie', 'matemat_session_id=; expires=Thu, 01 Jan 1970 00:00:00 GMT') - self.send_header('Location', '/') - self.end_headers() - return - self.session_id: str = session_id - - # Set all static headers. These can still be overwritten by a pagelet - headers: Dict[str, str] = dict() - for k, v in self.server.static_headers.items(): - headers[k] = v - - # Call a pagelet function, if one is registered for the requested path - if path in _PAGELET_PATHS: - # Call the pagelet function - pagelet_res = _PAGELET_PATHS[path](method, - path, - args, - self.session_vars, - headers, - self.server.pagelet_variables) - # Parse the pagelet's return value, vielding a HTTP status code and a response body - hsc, data = self._parse_pagelet_result(pagelet_res, headers) - # Send the HTTP status code - self.send_response(hsc) - # Format the session cookie timeout string and send the session cookie header - expires = timeout.strftime('%a, %d %b %Y %H:%M:%S GMT') - headers['Set-Cookie'] = f'matemat_session_id={session_id}; expires={expires}' - # Disable caching - headers['Cache-Control'] = 'no-cache' - # Compute the body length and add the appropriate header - headers['Content-Length'] = str(len(data)) - # If the pagelet did not set its own Content-Type header, use libmagic to guess an appropriate one - if 'Content-Type' not in headers: - try: - filemagic: magic.FileMagic = magic.detect_from_content(data) - mimetype: str = filemagic.mime_type - charset: str = filemagic.encoding - except ValueError: - mimetype = 'application/octet-stream' - charset = 'binary' - # Only append the charset if it is not "binary" - if charset == 'binary': - headers['Content-Type'] = mimetype - else: - headers['Content-Type'] = f'{mimetype}; charset={charset}' - # Send all headers set by the pagelet - for name, value in headers.items(): - self.send_header(name, value) - # End the header section and write the body - self.end_headers() - self.wfile.write(data) - else: - # No pagelet function for this path, try a static serve instead - # Only HTTP GET is allowed, else reply with a 'Method Not Allowed' header - if method != 'GET': - self.send_error(405) - self.end_headers() - return - # Create the absolute resource path, resolving '..' - filepath: str = os.path.abspath(os.path.join(self.server.webroot, path[1:])) - # Make sure the file is actually inside the webroot directory and that it exists - if os.path.commonpath([filepath, self.server.webroot]) == self.server.webroot and os.path.exists(filepath): - # Parse the If-Modified-Since header to check whether the browser can reuse cached content - datestr: str = self.headers.get('If-Modified-Since', 'Thu, 01 Jan 1970 00:00:00 GMT') - maxage: datetime = datetime.strptime(datestr, '%a, %d %b %Y %H:%M:%S %Z') - # Get file modification time - filestat: int = int(os.path.getmtime(filepath)) - # Create UTC datetime object from mtime - fileage: datetime = datetime.utcfromtimestamp(filestat) - - # If the file has not been replaced by a newer version than requested by the client, send a 304 response - if fileage <= maxage: - self.send_response(304, 'Not Modified') - self.send_header('Content-Length', '0') - self.end_headers() - return - - # Open and read the file - with open(filepath, 'rb') as f: - data = f.read() - # File read successfully, send 'OK' header - self.send_response(200) - # Guess the MIME type by file extension, or use libmagic as fallback - # Use libmagic to guess the charset - try: - exttype: str = mimetypes.guess_type(filepath)[0] - filemagic: magic.FileMagic = magic.detect_from_filename(filepath) - mimetype: str = exttype if exttype is not None else filemagic.mime_type - charset: str = filemagic.encoding - except ValueError: - mimetype = 'application/octet-stream' - charset = 'binary' - # Send content type and length header. Only set the charset if it's not "binary" - if charset == 'binary': - headers['Content-Type'] = mimetype - else: - headers['Content-Type'] = f'{mimetype}; charset={charset}' - headers['Content-Length'] = str(len(data)) - headers['Last-Modified'] = fileage.strftime('%a, %d %b %Y %H:%M:%S GMT') - headers['Cache-Control'] = 'max-age=1' - # Send all headers - for name, value in headers.items(): - self.send_header(name, value) - self.end_headers() - # Send the requested resource as response body - self.wfile.write(data) - else: - # File does not exist or path points outside the webroot directory - self.send_response(404) - self.end_headers() - - def _handle_request(self, method: str, path: str, args: RequestArguments): - try: - self._handle(method, path, args) - # Special handling for some errors - except HttpException as e: - self.send_error(e.status, e.title, e.message) - if 500 <= e.status < 600: - self.send_error(500, 'Internal Server Error') - self.server.logger.exception('', exc_info=e) - else: - self.server.logger.debug('', exc_info=e) - except PermissionError as e: - self.send_error(403, 'Forbidden') - self.server.logger.debug('', exc_info=e) - except ValueError as e: - self.send_error(400, 'Bad Request') - self.server.logger.debug('', exc_info=e) - except BaseException as e: - # Generic error handling - self.send_error(500, 'Internal Server Error') - self.server.logger.exception('', e.args, e) - - # noinspection PyPep8Naming - def do_GET(self) -> None: - """ - Called by BasicHTTPRequestHandler for GET requests. - """ - # Parse the request and hand it to the handle function - try: - path, args = parse_args(self.path) - self._handle_request('GET', path, args) - except ValueError as e: - self.send_error(400, 'Bad Request') - self.server.logger.debug('', exc_info=e) - - # noinspection PyPep8Naming - def do_POST(self) -> None: - """ - Called by BasicHTTPRequestHandler for POST requests. - """ - try: - # Read the POST body, if it exists, and its MIME type is application/x-www-form-urlencoded - clen: int = int(str(self.headers.get('Content-Length', failobj='0'))) - if clen > _MAX_POST: - # Return a 413 error page if the request size exceeds boundaries - self.send_error(413, 'Payload Too Large') - self.server.logger.debug('', exc_info=HttpException(413, 'Payload Too Large')) - return - ctype: str = self.headers.get('Content-Type', failobj='application/octet-stream') - post: bytes = self.rfile.read(clen) - # Parse the request and hand it to the handle function - path, args = parse_args(self.path, postbody=post, enctype=ctype) - self._handle_request('POST', path, args) - except ValueError as e: - self.send_error(400, 'Bad Request') - self.server.logger.debug('', exc_info=e) - - @property - def session_vars(self) -> Dict[str, Any]: - """ - Get the session variables for the current session. - - :return: Dictionary of named session variables. - """ - return self.server.session_vars[self.session_id][1] - - def log_request(self, code='-', size='-'): - self.server.logger.info('%s %s from %s', self.requestline, code, self.client_address) - - def log_error(self, fmt: str, *args): - self.server.logger.warning(f'{self.requestline} from {self.client_address}: {fmt}', *args) - - def log_message(self, fmt, *args): - self.server.logger.info(fmt, *args) diff --git a/matemat/webserver/logger.py b/matemat/webserver/logger.py new file mode 100644 index 0000000..a56eb2e --- /dev/null +++ b/matemat/webserver/logger.py @@ -0,0 +1,25 @@ +from typing import Any, Dict + +import logging + +from matemat.webserver.config import get_config + + +class Logger(logging.Logger): + + logger: 'Logger' = None + + def __init__(self): + super().__init__(name='matemat') + config: Dict[str, Any] = get_config() + log_handler: logging.Handler = config['log_handler'] + self.setLevel(config['log_level']) + # Set up the log handler's (obtained from config parsing) format string + log_handler.setFormatter(logging.Formatter('%(asctime)s %(module)s [%(levelname)s]: %(message)s')) + self.addHandler(log_handler) + + @classmethod + def instance(cls) -> 'Logger': + if cls.logger is None: + cls.logger = Logger() + return cls.logger diff --git a/matemat/webserver/pagelets/__init__.py b/matemat/webserver/pagelets/__init__.py index ae40e4d..bc4f3c4 100644 --- a/matemat/webserver/pagelets/__init__.py +++ b/matemat/webserver/pagelets/__init__.py @@ -4,7 +4,6 @@ This package contains the pagelet functions served by the Matemat software. A new pagelet function must be imported here to be automatically loaded when the server is started. """ -from .initialization import initialization from .main import main_page from .login import login_page from .logout import logout diff --git a/matemat/webserver/pagelets/admin.py b/matemat/webserver/pagelets/admin.py index 40046eb..1ad302c 100644 --- a/matemat/webserver/pagelets/admin.py +++ b/matemat/webserver/pagelets/admin.py @@ -1,69 +1,67 @@ -from typing import Any, Dict, Union - import os -from shutil import copyfile -import magic from io import BytesIO -from PIL import Image +from shutil import copyfile + +import magic +from PIL import Image +from bottle import get, post, abort, redirect, request, FormsDict -from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse -from matemat.util.currency_format import parse_chf from matemat.db import MatematDatabase from matemat.db.primitives import User, ReceiptPreference -from matemat.exceptions import AuthenticationError, DatabaseConsistencyError, HttpException +from matemat.exceptions import AuthenticationError, DatabaseConsistencyError +from matemat.util.currency_format import parse_chf +from matemat.webserver import session, template +from matemat.webserver.config import get_app_config -@pagelet('/admin') -def admin(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - config: Dict[str, str]) \ - -> Union[str, bytes, PageletResponse]: +@get('/admin') +@post('/admin') +def admin(): """ The admin panel, shows a user's own settings. Additionally, for administrators, settings to modify other users and products are shown. """ + config = get_app_config() + session_id: str = session.start() # If no user is logged in, redirect to the login page - if 'authentication_level' not in session_vars or 'authenticated_user' not in session_vars: - return RedirectResponse('/login') - authlevel: int = session_vars['authentication_level'] - uid: int = session_vars['authenticated_user'] + if not session.has(session_id, 'authentication_level') or not session.has(session_id, 'authenticated_user'): + redirect('/login') + authlevel: int = session.get(session_id, 'authentication_level') + uid: int = session.get(session_id, 'authenticated_user') # Show a 403 Forbidden error page if no user is logged in (0) or a user logged in via touchkey (1) if authlevel < 2: - raise HttpException(403) + abort(403) # Connect to the database with MatematDatabase(config['DatabaseFile']) as db: # Fetch the authenticated user user = db.get_user(uid) # If the POST request contains a "change" parameter, delegate the change handling to the function below - if method == 'POST' and 'change' in args: - handle_change(args, user, db, config) + if request.method == 'POST' and 'change' in request.params: + handle_change(request.params, request.files, user, db) # If the POST request contains an "adminchange" parameter, delegate the change handling to the function below - elif method == 'POST' and 'adminchange' in args and user.is_admin: - handle_admin_change(args, db, config) + elif request.method == 'POST' and 'adminchange' in request.params and user.is_admin: + handle_admin_change(request.params, request.files, db) # Fetch all existing users and products from the database users = db.list_users() products = db.list_products() # Render the "Admin/Settings" page - return TemplateResponse('admin.html', - authuser=user, authlevel=authlevel, users=users, products=products, - receipt_preference_class=ReceiptPreference, - setupname=config['InstanceName'], config_smtp_enabled=config['SmtpSendReceipts']) + return template.render('admin.html', + authuser=user, authlevel=authlevel, users=users, products=products, + receipt_preference_class=ReceiptPreference, + setupname=config['InstanceName'], config_smtp_enabled=config['SmtpSendReceipts']) -def handle_change(args: RequestArguments, user: User, db: MatematDatabase, config: Dict[str, str]) -> None: +def handle_change(args: FormsDict, files: FormsDict, user: User, db: MatematDatabase) -> None: """ Write the changes requested by a user for its own account to the database. - :param args: The RequestArguments object passed to the pagelet. + :param args: The FormsDict object passed to the pagelet. :param user: The user to edit. :param db: The database facade where changes are written to. - :param config: The dictionary of config file entries from the [Pagelets] section. """ + config = get_app_config() try: # Read the type of change requested by the user, then switch over it change = str(args.change) @@ -122,10 +120,10 @@ def handle_change(args: RequestArguments, user: User, db: MatematDatabase, confi # The user requested an avatar change elif change == 'avatar': # The new avatar field must be present - if 'avatar' not in args: + if 'avatar' not in files: return # Read the raw image data from the request - avatar = bytes(args.avatar) + avatar = files.avatar.file.read() # Only process the image, if its size is more than zero. Zero size means no new image was uploaded if len(avatar) == 0: return @@ -150,14 +148,14 @@ def handle_change(args: RequestArguments, user: User, db: MatematDatabase, confi raise ValueError('an argument not a string') -def handle_admin_change(args: RequestArguments, db: MatematDatabase, config: Dict[str, str]): +def handle_admin_change(args: FormsDict, files: FormsDict, db: MatematDatabase): """ Write the changes requested by an admin for users of products. :param args: The RequestArguments object passed to the pagelet. :param db: The database facade where changes are written to. - :param config: The dictionary of config file entries from the [Pagelets] section. """ + config = get_app_config() try: # Read the type of change requested by the admin, then switch over it change = str(args.adminchange) @@ -202,11 +200,10 @@ def handle_admin_change(args: RequestArguments, db: MatematDatabase, config: Dic # Create the user in the database newproduct = db.create_product(name, price_member, price_non_member) # If a new product image was uploaded, process it - if 'image' in args and len(bytes(args.image)) > 0: - # Read the raw image data from the request - avatar = bytes(args.image) + image = files.image.file.read() if 'image' in files else None + if image is None or len(image) == 0: # Detect the MIME type - filemagic: magic.FileMagic = magic.detect_from_content(avatar) + filemagic: magic.FileMagic = magic.detect_from_content(image) if not filemagic.mime_type.startswith('image/'): return # Create the absolute path of the upload directory @@ -214,7 +211,7 @@ def handle_admin_change(args: RequestArguments, db: MatematDatabase, config: Dic os.makedirs(abspath, exist_ok=True) try: # Parse the image data - image: Image = Image.open(BytesIO(avatar)) + image: Image = Image.open(BytesIO(image)) # Resize the image to 150x150 image.thumbnail((150, 150), Image.LANCZOS) # Write the image to the file @@ -253,7 +250,7 @@ def handle_admin_change(args: RequestArguments, db: MatematDatabase, config: Dic if category not in args: continue # Read the raw image data from the request - default: bytes = bytes(args[category]) + default: bytes = files[category].file.read() # Only process the image, if its size is more than zero. Zero size means no new image was uploaded if len(default) == 0: continue diff --git a/matemat/webserver/pagelets/buy.py b/matemat/webserver/pagelets/buy.py index d08bafa..db47ca0 100644 --- a/matemat/webserver/pagelets/buy.py +++ b/matemat/webserver/pagelets/buy.py @@ -1,33 +1,31 @@ -from typing import Any, Dict, Union +from bottle import get, post, redirect, request -from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse from matemat.db import MatematDatabase +from matemat.webserver import session +from matemat.webserver.config import get_app_config -@pagelet('/buy') -def buy(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - config: Dict[str, str]) \ - -> Union[str, bytes, PageletResponse]: +@get('/buy') +@post('/buy') +def buy(): """ The purchasing mechanism. Called by the user clicking an item on the product list. """ + config = get_app_config() + session_id: str = session.start() # If no user is logged in, redirect to the main page, as a purchase must always be bound to a user - if 'authenticated_user' not in session_vars: - return RedirectResponse('/') + if not session.has(session_id, 'authenticated_user'): + redirect('/') # Connect to the database with MatematDatabase(config['DatabaseFile']) as db: # Fetch the authenticated user from the database - uid: int = session_vars['authenticated_user'] + uid: int = session.get(session_id, 'authenticated_user') user = db.get_user(uid) # Read the product from the database, identified by the product ID passed as request argument - if 'pid' in args: - pid = int(str(args.pid)) + if 'pid' in request.params: + pid = int(str(request.params.pid)) product = db.get_product(pid) # Create a consumption entry for the (user, product) combination db.increment_consumption(user, product) # Redirect to the main page (where this request should have come from) - return RedirectResponse('/') + redirect('/') diff --git a/matemat/webserver/pagelets/deposit.py b/matemat/webserver/pagelets/deposit.py index 9e38a0b..2da8ea2 100644 --- a/matemat/webserver/pagelets/deposit.py +++ b/matemat/webserver/pagelets/deposit.py @@ -1,32 +1,30 @@ -from typing import Any, Dict, Union +from bottle import get, post, redirect, request -from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse from matemat.db import MatematDatabase +from matemat.webserver import session +from matemat.webserver.config import get_app_config -@pagelet('/deposit') -def deposit(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - config: Dict[str, str]) \ - -> Union[str, bytes, PageletResponse]: +@get('/deposit') +@post('/deposit') +def deposit(): """ The cash depositing mechanism. Called by the user submitting a deposit from the product list. """ + config = get_app_config() + session_id: str = session.start() # If no user is logged in, redirect to the main page, as a deposit must always be bound to a user - if 'authenticated_user' not in session_vars: - return RedirectResponse('/') + if not session.has(session_id, 'authenticated_user'): + redirect('/') # Connect to the database with MatematDatabase(config['DatabaseFile']) as db: # Fetch the authenticated user from the database - uid: int = session_vars['authenticated_user'] + uid: int = session.get(session_id, 'authenticated_user') user = db.get_user(uid) - if 'n' in args: + if 'n' in request.params: # Read the amount of cash to deposit from the request arguments - n = int(str(args.n)) + n = int(str(request.params.n)) # Write the deposit to the database db.deposit(user, n) # Redirect to the main page (where this request should have come from) - return RedirectResponse('/') + redirect('/') diff --git a/matemat/webserver/pagelets/initialization.py b/matemat/webserver/pagelets/initialization.py deleted file mode 100644 index fcb94d9..0000000 --- a/matemat/webserver/pagelets/initialization.py +++ /dev/null @@ -1,53 +0,0 @@ -from typing import Dict - -import logging - -from matemat.webserver import pagelet_init -from matemat.db import MatematDatabase - - -@pagelet_init -def initialization(config: Dict[str, str], - logger: logging.Logger) -> None: - """ - The pagelet initialization function. Makes sure everything is ready for operation. If anything fails here, the web - server won't resume operation. - """ - # Set default values for missing config items - if 'InstanceName' not in config: - config['InstanceName'] = 'Matemat' - logger.warning('Property \'InstanceName\' not set, using \'Matemat\'') - if 'UploadDir' not in config: - config['UploadDir'] = './static/upload/' - logger.warning('Property \'UploadDir\' not set, using \'./static/upload/\'') - if 'DatabaseFile' not in config: - config['DatabaseFile'] = './matemat.db' - logger.warning('Property \'DatabaseFile\' not set, using \'./matemat.db\'') - if 'SmtpSendReceipts' not in config: - config['SmtpSendReceipts'] = '0' - logger.warning('Property \'SmtpSendReceipts\' not set, using \'0\'') - if config['SmtpSendReceipts'] == '1': - if 'SmtpFrom' not in config: - logger.fatal('\'SmtpSendReceipts\' set to \'1\', but \'SmtpFrom\' missing.') - raise KeyError() - if 'SmtpSubj' not in config: - logger.fatal('\'SmtpSendReceipts\' set to \'1\', but \'SmtpSubj\' missing.') - raise KeyError() - if 'SmtpHost' not in config: - logger.fatal('\'SmtpSendReceipts\' set to \'1\', but \'SmtpHost\' missing.') - raise KeyError() - if 'SmtpPort' not in config: - logger.fatal('\'SmtpSendReceipts\' set to \'1\', but \'SmtpPort\' missing.') - raise KeyError() - if 'SmtpUser' not in config: - logger.fatal('\'SmtpSendReceipts\' set to \'1\', but \'SmtpUser\' missing.') - raise KeyError() - if 'SmtpPass' not in config: - logger.fatal('\'SmtpSendReceipts\' set to \'1\', but \'SmtpPass\' missing.') - raise KeyError() - if 'SmtpEnforceTLS' not in config: - config['SmtpEnforceTLS'] = '1' - logger.warning('Property \'SmtpEnforceTLS\' not set, using \'1\'') - with MatematDatabase(config['DatabaseFile']): - # Connect to the database to create it and perform any schema migrations - pass diff --git a/matemat/webserver/pagelets/login.py b/matemat/webserver/pagelets/login.py index be4ac4b..8ef2adb 100644 --- a/matemat/webserver/pagelets/login.py +++ b/matemat/webserver/pagelets/login.py @@ -1,45 +1,43 @@ -from typing import Any, Dict, Union +from bottle import get, post, redirect, abort, request -from matemat.exceptions import AuthenticationError, HttpException -from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse -from matemat.db.primitives import User from matemat.db import MatematDatabase +from matemat.db.primitives import User +from matemat.exceptions import AuthenticationError +from matemat.webserver import template, session +from matemat.webserver.config import get_app_config -@pagelet('/login') -def login_page(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - config: Dict[str, str]) \ - -> Union[bytes, str, PageletResponse]: +@get('/login') +@post('/login') +def login_page(): """ The password login mechanism. If called via GET, render the UI template; if called via POST, attempt to log in with the provided credentials (username and passsword). """ + config = get_app_config() + session_id: str = session.start() # If a user is already logged in, simply redirect to the main page, showing the product list - if 'authenticated_user' in session_vars: - return RedirectResponse('/') + if session.has(session_id, 'authenticated_user'): + redirect('/') # If requested via HTTP GET, render the login page showing the login UI - if method == 'GET': - return TemplateResponse('login.html', - setupname=config['InstanceName']) + if request.method == 'GET': + return template.render('login.html', + setupname=config['InstanceName']) # If requested via HTTP POST, read the request arguments and attempt to log in with the provided credentials - elif method == 'POST': + elif request.method == 'POST': # Connect to the database with MatematDatabase(config['DatabaseFile']) as db: try: # Read the request arguments and attempt to log in with them - user: User = db.login(str(args.username), str(args.password)) + user: User = db.login(str(request.params.username), str(request.params.password)) except AuthenticationError: # Reload the touchkey login page on failure - return RedirectResponse('/login') + redirect('/login') # Set the user ID session variable - session_vars['authenticated_user'] = user.id + session.put(session_id, 'authenticated_user', user.id) # Set the authlevel session variable (0 = none, 1 = touchkey, 2 = password login) - session_vars['authentication_level'] = 2 + session.put(session_id, 'authentication_level', 2) # Redirect to the main page, showing the product list - return RedirectResponse('/') + redirect('/') # If neither GET nor POST was used, show a 405 Method Not Allowed error page - raise HttpException(405) + abort(405) diff --git a/matemat/webserver/pagelets/logout.py b/matemat/webserver/pagelets/logout.py index 5d8f289..d4094bf 100644 --- a/matemat/webserver/pagelets/logout.py +++ b/matemat/webserver/pagelets/logout.py @@ -1,23 +1,19 @@ -from typing import Any, Dict, Union +from bottle import get, post, redirect -from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse +from matemat.webserver import session -@pagelet('/logout') -def logout(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - config: Dict[str, str]) \ - -> Union[bytes, str, PageletResponse]: +@get('/logout') +@post('/logout') +def logout(): """ The logout mechanism, clearing the authentication values in the session storage. """ + session_id: str = session.start() # Remove the authenticated user ID from the session storage, if any - if 'authenticated_user' in session_vars: - del session_vars['authenticated_user'] + if session.has(session_id, 'authenticated_user'): + session.delete(session_id, 'authenticated_user') # Reset the authlevel session variable (0 = none, 1 = touchkey, 2 = password login) - session_vars['authentication_level'] = 0 + session.put(session_id, 'authentication_level', 0) # Redirect to the main page, showing the user list - return RedirectResponse('/') + redirect('/') diff --git a/matemat/webserver/pagelets/main.py b/matemat/webserver/pagelets/main.py index 54f6f79..49ddd80 100644 --- a/matemat/webserver/pagelets/main.py +++ b/matemat/webserver/pagelets/main.py @@ -1,39 +1,36 @@ -from typing import Any, Dict, Union +from bottle import route, redirect -from matemat.webserver import pagelet, RequestArguments, PageletResponse, TemplateResponse, RedirectResponse from matemat.db import MatematDatabase +from matemat.webserver import template, session +from matemat.webserver.config import get_app_config -@pagelet('/') -def main_page(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - config: Dict[str, str]) \ - -> Union[bytes, str, PageletResponse]: +@route('/') +def main_page(): """ The main page, showing either the user list (if no user is logged in) or the product list (if a user is logged in). """ + config = get_app_config() + session_id: str = session.start() with MatematDatabase(config['DatabaseFile']) as db: # Check whether a user is logged in - if 'authenticated_user' in session_vars: + if session.has(session_id, 'authenticated_user'): # Fetch the user id and authentication level (touchkey vs password) from the session storage - uid: int = session_vars['authenticated_user'] - authlevel: int = session_vars['authentication_level'] + uid: int = session.get(session_id, 'authenticated_user') + authlevel: int = session.get(session_id, 'authentication_level') # Fetch the user object from the database (for name display, price calculation and admin check) user = db.get_user(uid) # Fetch the list of products to display products = db.list_products() # Prepare a response with a jinja2 template - return TemplateResponse('productlist.html', - authuser=user, products=products, authlevel=authlevel, - setupname=config['InstanceName']) + return template.render('productlist.html', + authuser=user, products=products, authlevel=authlevel, + setupname=config['InstanceName']) else: # If there are no admin users registered, jump to the admin creation procedure if not db.has_admin_users(): - return RedirectResponse('/userbootstrap') + redirect('/userbootstrap') # If no user is logged in, fetch the list of users and render the userlist template users = db.list_users(with_touchkey=True) - return TemplateResponse('userlist.html', - users=users, setupname=config['InstanceName']) + return template.render('userlist.html', + users=users, setupname=config['InstanceName']) diff --git a/matemat/webserver/pagelets/modproduct.py b/matemat/webserver/pagelets/modproduct.py index 7ba9c3e..6139eaf 100644 --- a/matemat/webserver/pagelets/modproduct.py +++ b/matemat/webserver/pagelets/modproduct.py @@ -1,36 +1,35 @@ -from typing import Any, Dict, Union - import os +from io import BytesIO +from typing import Dict + import magic from PIL import Image -from io import BytesIO +from bottle import get, post, redirect, abort, request, FormsDict -from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse from matemat.db import MatematDatabase from matemat.db.primitives import Product -from matemat.exceptions import DatabaseConsistencyError, HttpException +from matemat.exceptions import DatabaseConsistencyError from matemat.util.currency_format import parse_chf +from matemat.webserver import template, session +from matemat.webserver.config import get_app_config -@pagelet('/modproduct') -def modproduct(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - config: Dict[str, str]) \ - -> Union[str, bytes, PageletResponse]: +@get('/modproduct') +@post('/modproduct') +def modproduct(): """ The product modification page available from the admin panel. """ + config = get_app_config() + session_id: str = session.start() # If no user is logged in, redirect to the login page - if 'authentication_level' not in session_vars or 'authenticated_user' not in session_vars: - return RedirectResponse('/login') - authlevel: int = session_vars['authentication_level'] - auth_uid: int = session_vars['authenticated_user'] + if not session.has(session_id, 'authentication_level') or not session.has(session_id, 'authenticated_user'): + redirect('/login') + authlevel: int = session.get(session_id, 'authentication_level') + auth_uid: int = session.get(session_id, 'authenticated_user') # Show a 403 Forbidden error page if no user is logged in (0) or a user logged in via touchkey (1) if authlevel < 2: - raise HttpException(403) + abort(403) # Connect to the database with MatematDatabase(config['DatabaseFile']) as db: @@ -38,38 +37,38 @@ def modproduct(method: str, authuser = db.get_user(auth_uid) if not authuser.is_admin: # Show a 403 Forbidden error page if the user is not an admin - raise HttpException(403) - if 'productid' not in args: + abort(403) + if 'productid' not in request.params: # Show a 400 Bad Request error page if no product to edit was specified # (should never happen during normal operation) - raise HttpException(400, '"productid" argument missing') + abort(400, '"productid" argument missing') # Fetch the product to modify from the database - modproduct_id = int(str(args.productid)) + modproduct_id = int(str(request.params.productid)) product = db.get_product(modproduct_id) # If the request contains a "change" parameter, delegate the change handling to the function below - if 'change' in args: - handle_change(args, product, db, config) + if 'change' in request.params: + handle_change(request.params, request.files, product, db) # If the product was deleted, redirect back to the admin page, as there is nothing to edit any more - if str(args.change) == 'del': - return RedirectResponse('/admin') + if str(request.params.change) == 'del': + redirect('/admin') # Render the "Modify Product" page - return TemplateResponse('modproduct.html', - authuser=authuser, product=product, authlevel=authlevel, - setupname=config['InstanceName']) + return template.render('modproduct.html', + authuser=authuser, product=product, authlevel=authlevel, + setupname=config['InstanceName']) -def handle_change(args: RequestArguments, product: Product, db: MatematDatabase, config: Dict[str, str]) -> None: +def handle_change(args: FormsDict, files: FormsDict, product: Product, db: MatematDatabase) -> None: """ Write the changes requested by an admin to the database. - :param args: The RequestArguments object passed to the pagelet. + :param args: The FormsDict object passed to the pagelet. :param product: The product to edit. :param db: The database facade where changes are written to. - :param config: The dictionary of config file entries from the [Pagelets] section. """ + config = get_app_config() # Read the type of change requested by the admin, then switch over it change = str(args.change) @@ -101,9 +100,9 @@ def handle_change(args: RequestArguments, product: Product, db: MatematDatabase, except DatabaseConsistencyError: return # If a new product image was uploaded, process it - if 'image' in args: + if 'image' in files: # Read the raw image data from the request - avatar = bytes(args.image) + avatar = files.image.file.read() # Only process the image, if its size is more than zero. Zero size means no new image was uploaded if len(avatar) == 0: return diff --git a/matemat/webserver/pagelets/moduser.py b/matemat/webserver/pagelets/moduser.py index 94bfa84..0e5546b 100644 --- a/matemat/webserver/pagelets/moduser.py +++ b/matemat/webserver/pagelets/moduser.py @@ -1,36 +1,35 @@ -from typing import Any, Dict, Optional, Union - import os +from io import BytesIO +from typing import Dict, Optional + import magic from PIL import Image -from io import BytesIO +from bottle import get, post, redirect, abort, request, FormsDict -from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse from matemat.db import MatematDatabase from matemat.db.primitives import User, ReceiptPreference -from matemat.exceptions import DatabaseConsistencyError, HttpException +from matemat.exceptions import DatabaseConsistencyError from matemat.util.currency_format import parse_chf +from matemat.webserver import template, session +from matemat.webserver.config import get_app_config -@pagelet('/moduser') -def moduser(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - config: Dict[str, str]) \ - -> Union[str, bytes, PageletResponse]: +@get('/moduser') +@post('/moduser') +def moduser(): """ The user modification page available from the admin panel. """ + config = get_app_config() + session_id: str = session.start() # If no user is logged in, redirect to the login page - if 'authentication_level' not in session_vars or 'authenticated_user' not in session_vars: - return RedirectResponse('/login') - authlevel: int = session_vars['authentication_level'] - auth_uid: int = session_vars['authenticated_user'] + if not session.has(session_id, 'authentication_level') or not session.has(session_id, 'authenticated_user'): + redirect('/login') + authlevel: int = session.get(session_id, 'authentication_level') + auth_uid: int = session.get(session_id, 'authenticated_user') # Show a 403 Forbidden error page if no user is logged in (0) or a user logged in via touchkey (1) if authlevel < 2: - raise HttpException(403) + abort(403) # Connect to the database with MatematDatabase(config['DatabaseFile']) as db: @@ -38,31 +37,31 @@ def moduser(method: str, authuser = db.get_user(auth_uid) if not authuser.is_admin: # Show a 403 Forbidden error page if the user is not an admin - raise HttpException(403) - if 'userid' not in args: + abort(403) + if 'userid' not in request.params: # Show a 400 Bad Request error page if no users to edit was specified # (should never happen during normal operation) - raise HttpException(400, '"userid" argument missing') + abort(400, '"userid" argument missing') # Fetch the user to modify from the database - moduser_id = int(str(args.userid)) + moduser_id = int(str(request.params.userid)) user = db.get_user(moduser_id) # If the request contains a "change" parameter, delegate the change handling to the function below - if 'change' in args: - handle_change(args, user, authuser, db, config) + if 'change' in request.params: + handle_change(request.params, request.files, user, authuser, db) # If the user was deleted, redirect back to the admin page, as there is nothing to edit any more - if str(args.change) == 'del': - return RedirectResponse('/admin') + if str(request.params.change) == 'del': + redirect('/admin') # Render the "Modify User" page - return TemplateResponse('moduser.html', - authuser=authuser, user=user, authlevel=authlevel, - receipt_preference_class=ReceiptPreference, - setupname=config['InstanceName'], config_smtp_enabled=config['SmtpSendReceipts']) + return template.render('moduser.html', + authuser=authuser, user=user, authlevel=authlevel, + receipt_preference_class=ReceiptPreference, + setupname=config['InstanceName'], config_smtp_enabled=config['SmtpSendReceipts']) -def handle_change(args: RequestArguments, user: User, authuser: User, db: MatematDatabase, config: Dict[str, str]) \ +def handle_change(args: FormsDict, files: FormsDict, user: User, authuser: User, db: MatematDatabase) \ -> None: """ Write the changes requested by an admin to the database. @@ -71,8 +70,8 @@ def handle_change(args: RequestArguments, user: User, authuser: User, db: Matema :param user: The user to edit. :param authuser: The user performing the modification. :param db: The database facade where changes are written to. - :param config: The dictionary of config file entries from the [Pagelets] section. """ + config = get_app_config() # Read the type of change requested by the admin, then switch over it change = str(args.change) @@ -124,9 +123,9 @@ def handle_change(args: RequestArguments, user: User, authuser: User, db: Matema except DatabaseConsistencyError: return # If a new avatar was uploaded, process it - if 'avatar' in args: + if 'avatar' in files: # Read the raw image data from the request - avatar = bytes(args.avatar) + avatar = files.avatar.file.read() # Only process the image, if its size is more than zero. Zero size means no new image was uploaded if len(avatar) == 0: return diff --git a/matemat/webserver/pagelets/receipt_smtp_cron.py b/matemat/webserver/pagelets/receipt_smtp_cron.py index 389cab1..dc64479 100644 --- a/matemat/webserver/pagelets/receipt_smtp_cron.py +++ b/matemat/webserver/pagelets/receipt_smtp_cron.py @@ -1,22 +1,25 @@ -from typing import Dict, List, Tuple +from typing import List, Tuple import logging import smtplib as smtp from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText -from jinja2 import Environment, Template -from matemat.webserver import pagelet_cron +from matemat.webserver.cron import cron +from matemat.webserver import template +from matemat.webserver.config import get_app_config from matemat.db import MatematDatabase from matemat.db.primitives import User, Receipt from matemat.util.currency_format import format_chf -@pagelet_cron(hours=6) -def receipt_smtp_cron(config: Dict[str, str], - jinja_env: Environment, - logger: logging.Logger) -> None: +logger: logging.Logger = logging.Logger(__file__) + + +@cron(hours=6) +def receipt_smtp_cron() -> None: + config = get_app_config() if config['SmtpSendReceipts'] != '1': # Sending receipts via mail is disabled return @@ -35,15 +38,13 @@ def receipt_smtp_cron(config: Dict[str, str], receipts.append(receipt) else: logger.debug('No receipt due for user "%s".', user.name) - # Send all generated receipts via e-mailgi + # Send all generated receipts via e-mail if len(receipts) > 0: - _send_receipt_mails(receipts, jinja_env, logger, config) + _send_receipt_mails(receipts) -def _send_receipt_mails(receipts: List[Receipt], - jinja_env: Environment, - logger: logging.Logger, - config: Dict[str, str]) -> None: +def _send_receipt_mails(receipts: List[Receipt]) -> None: + config = get_app_config() mails: List[Tuple[str, MIMEMultipart]] = [] for receipt in receipts: if receipt.user.email is None: @@ -63,8 +64,8 @@ def _send_receipt_mails(receipts: List[Receipt], fbal = format_chf(receipt.transactions[0].old_balance).rjust(12) tbal: str = format_chf(receipt.user.balance).rjust(12) # Render the receipt - template: Template = jinja_env.get_template('receipt.txt') - rendered: str = template.render(fdate=fdate, tdate=tdate, user=username, fbal=fbal, tbal=tbal, + rendered: str = template.render('receipt.txt', + date=fdate, tdate=tdate, user=username, fbal=fbal, tbal=tbal, receipt_id=receipt.id, transactions=receipt.transactions, instance_name=config['InstanceName']) # Put the rendered receipt in the message body diff --git a/matemat/webserver/pagelets/statistics.py b/matemat/webserver/pagelets/statistics.py index 61aef7c..7ff6a11 100644 --- a/matemat/webserver/pagelets/statistics.py +++ b/matemat/webserver/pagelets/statistics.py @@ -1,33 +1,30 @@ -from typing import Any, Dict, List, Tuple, Union - -from math import pi, sin, cos from datetime import datetime, timedelta +from math import pi, sin, cos +from typing import Any, Dict, List, Tuple + +from bottle import route, redirect, abort, request -from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse from matemat.db import MatematDatabase from matemat.db.primitives import User -from matemat.exceptions import HttpException +from matemat.webserver import template, session +from matemat.webserver.config import get_app_config -@pagelet('/statistics') -def statistics(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - config: Dict[str, str]) \ - -> Union[str, bytes, PageletResponse]: +@route('/statistics') +def statistics(): """ The statistics page available from the admin panel. """ + config = get_app_config() + session_id: str = session.start() # If no user is logged in, redirect to the login page - if 'authentication_level' not in session_vars or 'authenticated_user' not in session_vars: - return RedirectResponse('/login') - authlevel: int = session_vars['authentication_level'] - auth_uid: int = session_vars['authenticated_user'] + if not session.has(session_id, 'authentication_level') or not session.has(session_id, 'authenticated_user'): + redirect('/login') + authlevel: int = session.get(session_id, 'authentication_level') + auth_uid: int = session.get(session_id, 'authenticated_user') # Show a 403 Forbidden error page if no user is logged in (0) or a user logged in via touchkey (1) if authlevel < 2: - raise HttpException(403) + abort(403) # Connect to the database with MatematDatabase(config['DatabaseFile']) as db: @@ -35,15 +32,15 @@ def statistics(method: str, authuser: User = db.get_user(auth_uid) if not authuser.is_admin: # Show a 403 Forbidden error page if the user is not an admin - raise HttpException(403) + abort(403) todate: datetime = datetime.utcnow() fromdate: datetime = (todate - timedelta(days=365)).replace(hour=0, minute=0, second=0, microsecond=0) - if 'fromdate' in args: - fdarg: str = str(args.fromdate) + if 'fromdate' in request.params: + fdarg: str = str(request.params.fromdate) fromdate = datetime.strptime(fdarg, '%Y-%m-%d').replace(tzinfo=None) - if 'todate' in args: - tdarg: str = str(args.todate) + if 'todate' in request.params: + tdarg: str = str(request.params.todate) todate = datetime.strptime(tdarg, '%Y-%m-%d').replace(tzinfo=None) todate = (todate + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) @@ -70,10 +67,10 @@ def statistics(method: str, previous_angle = angle # Render the statistics page - return TemplateResponse('statistics.html', - fromdate=fromdate.strftime('%Y-%m-%d'), - todate=(todate - timedelta(days=1)).strftime('%Y-%m-%d'), - product_slices=slices, - authuser=authuser, authlevel=authlevel, - setupname=config['InstanceName'], - **stats) + return template.render('statistics.html', + fromdate=fromdate.strftime('%Y-%m-%d'), + todate=(todate - timedelta(days=1)).strftime('%Y-%m-%d'), + product_slices=slices, + authuser=authuser, authlevel=authlevel, + setupname=config['InstanceName'], + **stats) diff --git a/matemat/webserver/pagelets/touchkey.py b/matemat/webserver/pagelets/touchkey.py index 69f5b8e..9b5fd87 100644 --- a/matemat/webserver/pagelets/touchkey.py +++ b/matemat/webserver/pagelets/touchkey.py @@ -1,46 +1,44 @@ -from typing import Any, Dict, Union +from bottle import get, post, redirect, abort, request -from matemat.exceptions import AuthenticationError, HttpException -from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse -from matemat.db.primitives import User from matemat.db import MatematDatabase +from matemat.db.primitives import User +from matemat.exceptions import AuthenticationError +from matemat.webserver import template, session +from matemat.webserver.config import get_app_config -@pagelet('/touchkey') -def touchkey_page(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - config: Dict[str, str]) \ - -> Union[bytes, str, PageletResponse]: +@get('/touchkey') +@post('/touchkey') +def touchkey_page(): """ The touchkey login mechanism. If called via GET, render the UI template; if called via POST, attempt to log in with the provided credentials (username and touchkey). """ + config = get_app_config() + session_id: str = session.start() # If a user is already logged in, simply redirect to the main page, showing the product list - if 'authenticated_user' in session_vars: - return RedirectResponse('/') + if session.has(session_id, 'authenticated_user'): + redirect('/') # If requested via HTTP GET, render the login page showing the touchkey UI - if method == 'GET': - return TemplateResponse('touchkey.html', - username=str(args.username), uid=int(str(args.uid)), - setupname=config['InstanceName']) + if request.method == 'GET': + return template.render('touchkey.html', + username=str(request.params.username), uid=int(str(request.params.uid)), + setupname=config['InstanceName']) # If requested via HTTP POST, read the request arguments and attempt to log in with the provided credentials - elif method == 'POST': + elif request.method == 'POST': # Connect to the database with MatematDatabase(config['DatabaseFile']) as db: try: # Read the request arguments and attempt to log in with them - user: User = db.login(str(args.username), touchkey=str(args.touchkey)) + user: User = db.login(str(request.params.username), touchkey=str(request.params.touchkey)) except AuthenticationError: # Reload the touchkey login page on failure - return RedirectResponse(f'/touchkey?uid={str(args.uid)}&username={str(args.username)}') + redirect(f'/touchkey?uid={str(request.params.uid)}&username={str(request.params.username)}') # Set the user ID session variable - session_vars['authenticated_user'] = user.id + session.put(session_id, 'authenticated_user', user.id) # Set the authlevel session variable (0 = none, 1 = touchkey, 2 = password login) - session_vars['authentication_level'] = 1 + session.put(session_id, 'authentication_level', 1) # Redirect to the main page, showing the product list - return RedirectResponse('/') + redirect('/') # If neither GET nor POST was used, show a 405 Method Not Allowed error page - raise HttpException(405) + abort(405) diff --git a/matemat/webserver/pagelets/userbootstrap.py b/matemat/webserver/pagelets/userbootstrap.py index 8efdb39..2dbbaf5 100644 --- a/matemat/webserver/pagelets/userbootstrap.py +++ b/matemat/webserver/pagelets/userbootstrap.py @@ -1,43 +1,38 @@ - -from typing import Any, Dict, Union +from bottle import get, post, redirect, abort, request from matemat.db import MatematDatabase -from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse -from matemat.exceptions import HttpException +from matemat.webserver import template +from matemat.webserver.config import get_app_config -@pagelet('/userbootstrap') -def userbootstrap(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - config: Dict[str, str]) \ - -> Union[bytes, str, PageletResponse]: +@get('/userbootstrap') +@post('/userbootstrap') +def userbootstrap(): """ The page for creating a first admin user. To be used when the system is set up the first time, or when there are no admin users left. """ + config = get_app_config() with MatematDatabase(config['DatabaseFile']) as db: # Redirect to main if there are still administrators registered if db.has_admin_users(): - return RedirectResponse('/') + redirect('/') # Process submission - if method == 'POST': + if request.method == 'POST': # Make sure all required values are present - if 'username' not in args or 'password' not in args or 'password2' not in args: - raise HttpException(400, 'Some arguments are missing') - username: str = str(args.username) - password: str = str(args.password) - password2: str = str(args.password2) + if 'username' not in request.params or 'password' not in request.params or 'password2' not in request.params: + abort(400, 'Some arguments are missing') + username: str = str(request.params.username) + password: str = str(request.params.password) + password2: str = str(request.params.password2) # The 2 passwords must match if password != password2: - return RedirectResponse('/userbootstrap') + redirect('/userbootstrap') # Create the admin user db.create_user(username, password, None, True, False) # Redirect to the main page - return RedirectResponse('/') + redirect('/') # Requested via GET; show the user creation UI else: - return TemplateResponse('userbootstrap.html', - setupname=config['InstanceName']) + return template.render('userbootstrap.html', + setupname=config['InstanceName']) diff --git a/matemat/webserver/requestargs.py b/matemat/webserver/requestargs.py deleted file mode 100644 index 373db90..0000000 --- a/matemat/webserver/requestargs.py +++ /dev/null @@ -1,324 +0,0 @@ - -from typing import Dict, Iterator, List, Tuple, Union - - -class RequestArguments(object): - """ - Container for HTTP Request arguments. - - Usage: - - # Create empty instance - ra = RequestArguments() - # Add an entry for the key 'foo' with the value 'bar' and Content-Type 'text/plain' - ra['foo'].append('text/plain', 'bar') - # Retrieve the value for the key 'foo', as a string... - foo = str(ra.foo) - # ... or as raw bytes - foo = bytes(ra.foo) - """ - - def __init__(self) -> None: - """ - Create an empty container instance. - """ - self.__container: Dict[str, RequestArgument] = dict() - - def __getitem__(self, key: str) -> 'RequestArgument': - """ - Retrieve the argument for the given name, creating it on the fly, if it doesn't exist. - - :param key: Name of the argument to retrieve. - :return: A RequestArgument instance. - :raises TypeError: If key is not a string. - """ - if not isinstance(key, str): - raise TypeError('key must be a str') - # Create empty argument, if it doesn't exist - if key not in self.__container: - self.__container[key] = RequestArgument(key) - # Return the argument for the name - return self.__container[key] - - def __getattr__(self, key: str) -> 'RequestArgument': - """ - Syntactic sugar for accessing values with a name that can be used in Python attributes. The value will be - returned as an immutable view. - - :param key: Name of the argument to retrieve. - :return: An immutable view of the RequestArgument instance. - """ - return _View.of(self.__container[key]) - - def __iter__(self) -> Iterator['RequestArgument']: - """ - Returns an iterator over the values in this instance. Values are represented as immutable views. - - :return: An iterator that yields immutable views of the values. - """ - for ra in self.__container.values(): - # Yield an immutable scalar view for each value - yield _View.of(ra) - - def __contains__(self, key: str) -> bool: - """ - Checks whether an argument with a given name exists in the RequestArguments instance. - - :param key: The name to check whether it exists. - :return: True, if present, False otherwise. - """ - return key in self.__container - - def __len__(self) -> int: - """ - :return: The number of arguments in this instance. - """ - return len(self.__container) - - -class RequestArgument(object): - """ - Container class for HTTP request arguments that simplifies dealing with - - scalar and array arguments: - Automatically converts between single values and arrays where necessary: Arrays with one element can be - accessed as scalars, and scalars can be iterated, yielding themselves as a single item. - - UTF-8 strings and binary data (e.g. file uploads): - All data can be retrieved both as a str (if utf-8 decoding is possible) and a bytes object. - - The objects returned from iteration or indexing are immutable views of (parts of) this object. - - Usage example: - - qsargs = urllib.parse.parse_qs(qs, strict_parsing=True, keep_blank_values=True, errors='strict') - args: RequestArguments - for k, vs in qsargs: - args[k].clear() - for v in vs: - # text/plain usually is a sensible choice for values decoded from urlencoded strings - # IF ALREADY IN STRING FORM (which parse_qs does)! - args[k].append('text/plain', v) - - if 'username' in args and args.username.is_scalar: - username = str(args.username) - - """ - - def __init__(self, - name: str, - value: Union[Tuple[str, Union[bytes, str]], List[Tuple[str, Union[bytes, str]]]] = None) -> None: - """ - Create a new RequestArgument with a name and optionally an initial value. - - :param name: The name for this argument, as provided via GET or POST. - :param value: The initial value, if any. Optional, initializes with empty array if omitted. - """ - # Assign name - self.__name: str = name - # Initialize value - self.__value: List[Tuple[str, Union[bytes, str]]] = [] - # Default to empty array - if value is None: - self.__value = [] - else: - if isinstance(value, list): - # Store the array - self.__value = value - else: - # Turn scalar into an array before storing - self.__value = [value] - - @property - def is_array(self) -> bool: - """ - :return: True, if the value is a (possibly empty) array, False otherwise. - """ - return len(self.__value) != 1 - - @property - def is_scalar(self) -> bool: - """ - :return: True, if the value is a single scalar value, False otherwise. - """ - return len(self.__value) == 1 - - @property - def is_view(self) -> bool: - """ - :return: True, if this instance is an immutable view, False otherwise. - """ - return False - - @property - def name(self) -> str: - """ - :return: The name of this argument. - """ - return self.__name - - def get_str(self, index: int = 0) -> str: - """ - Attempts to return a value as a string. The index defaults to 0. - - :param index: The index of the value to retrieve. Default: 0. - :return: An UTF-8 string representation of the requested value. - :raises UnicodeDecodeError: If the value cannot be decoded into an UTF-8 string. - :raises IndexError: If the index is out of bounds. - :raises TypeError: If the index is not an int. - :raises TypeError: If the requested value is neither a str nor a bytes object. - """ - if not isinstance(index, int): - # Index must be an int - raise TypeError('index must be an int') - # Type hint; access array element - v: Tuple[str, Union[bytes, str]] = self.__value[index] - if isinstance(v[1], str): - # The value already is a string, return - return v[1] - elif isinstance(v[1], bytes): - # The value is a bytes object, attempt to decode - return v[1].decode('utf-8') - raise TypeError('Value is neither a str nor bytes') - - def __str__(self) -> str: - """ - Attempts to return the first value as a string. - :return: An UTF-8 string representation of the first value. - :raises UnicodeDecodeError: If the value cannot be decoded into an UTF-8 string. - """ - return self.get_str() - - def get_bytes(self, index: int = 0) -> bytes: - """ - Attempts to return a value as a bytes object. The index defaults to 0. - - :param index: The index of the value to retrieve. Default: 0. - :return: A bytes object representation of the requested value. Strings will be encoded as UTF-8. - :raises IndexError: If the index is out of bounds. - :raises TypeError: If the index is not an int. - :raises TypeError: If the requested value is neither a str nor a bytes object. - """ - if not isinstance(index, int): - # Index must be a int - raise TypeError('index must be an int') - # Type hint; access array element - v: Tuple[str, Union[bytes, str]] = self.__value[index] - if isinstance(v[1], bytes): - # The value already is a bytes object, return - return v[1] - elif isinstance(v[1], str): - # The value is a string, encode first - return v[1].encode('utf-8') - raise TypeError('Value is neither a str nor bytes') - - def __bytes__(self) -> bytes: - """ - Attempts to return the first value as a bytes object. - :return: A bytes string representation of the first value. - """ - return self.get_bytes() - - def get_content_type(self, index: int = 0) -> str: - """ - Attempts to retrieve a value's Content-Type. The index defaults to 0. - - :param index: The index of the value to retrieve. Default: 0. - :return: The Content-Type of the requested value, as sent by the client. Not necessarily trustworthy. - :raises IndexError: If the index is out of bounds. - :raises TypeError: If the index is not an int. - """ - # instance is an array value - if not isinstance(index, int): - # Needs an index for array values - raise TypeError('index must be an int') - # Type hint; access array element - va: Tuple[str, Union[bytes, str]] = self.__value[index] - # Return the content type of the requested value - if not isinstance(va[0], str): - raise TypeError('Content-Type is not a str') - return va[0] - - def append(self, ctype: str, value: Union[str, bytes]) -> None: - """ - Append a value to this instance. Turns an empty argument into a scalar and a scalar into an array. - - :param ctype: The Content-Type, as provided in the request. - :param value: The scalar value to append, either a string or bytes object. - :raises TypeError: If called on an immutable view. - """ - if self.is_view: - # This is an immutable view, raise exception - raise TypeError('A RequestArgument view is immutable!') - self.__value.append((ctype, value)) - - def clear(self) -> None: - """ - Remove all values from this instance. - - :raises TypeError: If called on an immutable view. - """ - if self.is_view: - # This is an immutable view, raise exception - raise TypeError('A RequestArgument view is immutable!') - self.__value.clear() - - def __len__(self) -> int: - """ - :return: Number of values for this argument. - """ - return len(self.__value) - - def __iter__(self) -> Iterator['RequestArgument']: - """ - Iterate the values of this argument. Each value is accessible as if it were a scalar RequestArgument in turn, - although they are immutable. - - :return: An iterator that yields immutable views of the values. - """ - for v in self.__value: - # Yield an immutable scalar view for each (ctype, value) element in the array - yield _View(self.__name, v) - - def __getitem__(self, index: Union[int, slice]) -> 'RequestArgument': - """ - Index the argument with either an int or a slice. The returned values are represented as immutable - RequestArgument views. - - :param index: The index or slice. - :return: An immutable view of the indexed elements of this argument. - """ - # Pass the index or slice through to the array, packing the result in an immutable view - return _View(self.__name, self.__value[index]) - - -class _View(RequestArgument): - """ - This class represents an immutable view of a (subset of a) RequestArgument object. Should not be instantiated - directly. - """ - - def __init__(self, name: str, value: Union[Tuple[str, Union[bytes, str]], List[Tuple[str, Union[bytes, str]]]])\ - -> None: - """ - Create a new immutable view of a (subset of a) RequestArgument. - - :param name: The name for this argument, same as in the original RequestArgument. - :param value: The values to represent in this view, obtained by e.g. indexing or slicing. - """ - super().__init__(name, value) - - @staticmethod - def of(argument: 'RequestArgument') ->'RequestArgument': - """ - Create an immutable, unsliced view of an RequestArgument instance. - - :param argument: The RequestArgument instance to create a view of. - :return: An immutable view of the provided RequestArgument instance. - """ - return argument[:] - - @property - def is_view(self) -> bool: - """ - :return: True, if this instance is an immutable view, False otherwise. - """ - return True diff --git a/matemat/webserver/responses.py b/matemat/webserver/responses.py deleted file mode 100644 index cf9dbf8..0000000 --- a/matemat/webserver/responses.py +++ /dev/null @@ -1,65 +0,0 @@ - -from jinja2 import Environment, Template - -from matemat import __version__ - - -class PageletResponse: - """ - Base class for pagelet return values that require more action than simply sending plain data. - - An instance of this base class will result in an empty 200 OK response. - """ - - def __init__(self, status: int = 200) -> None: - """ - Create an empty response. - - :param status: The HTTP status code, defaults to 200 (OK). - """ - self.status: int = status - - -class RedirectResponse(PageletResponse): - """ - A pagelet response that causes the server to redirect to another location, using a 301 Permanently Moved (uncached) - response status, and a Location header. - """ - - def __init__(self, location: str) -> None: - """ - Create a redirection response with the given redirection location. - - :param location: The location to redirect to. - """ - super().__init__(status=301) - self.location: str = location - - -class TemplateResponse(PageletResponse): - """ - A pagelet response that causes the server to load a Jinja2 template and render it with the provided arguments, then - sending the result as response body, with a 200 OK response status. - """ - - def __init__(self, name: str, **kwargs) -> None: - """ - Create a template response with the given template name and arguments. - - :param name: Name of the template to load. - :param kwargs: Arguments for rendering the template, will be passed to jinja2.Template.render as is. - """ - super().__init__() - self.name: str = name - self.kwargs = kwargs - - def _render(self, jinja_env: Environment) -> bytes: - """ - Load and render the template using the Jinja2 environment managed by the web server instance. This method - should not be called by a pagelet. - - :param jinja_env: The Jinja2 environment. - :return: An UTF-8 encoded bytes object containing the template rendering result. - """ - template: Template = jinja_env.get_template(self.name) - return template.render(**self.kwargs, __version__=__version__).encode('utf-8') diff --git a/matemat/webserver/session/__init__.py b/matemat/webserver/session/__init__.py new file mode 100644 index 0000000..b8aaacf --- /dev/null +++ b/matemat/webserver/session/__init__.py @@ -0,0 +1,2 @@ + +from .sessions import start, end, put, get, has, delete diff --git a/matemat/webserver/session/sessions.py b/matemat/webserver/session/sessions.py new file mode 100644 index 0000000..c1ee82c --- /dev/null +++ b/matemat/webserver/session/sessions.py @@ -0,0 +1,73 @@ +from typing import Any, Dict, Tuple, Optional + +from bottle import request, response +from secrets import token_bytes +from uuid import uuid4 +from datetime import datetime, timedelta + +__key: Optional[str] = token_bytes(32) + +__session_vars: Dict[str, Tuple[datetime, Dict[str, Any]]] = dict() + +# Inactivity timeout for client sessions +_SESSION_TIMEOUT: int = 3600 +_COOKIE_NAME: str = 'matemat_session_id' + + +def start() -> str: + """ + Start a new session, or resume the session identified by the session cookie sent in the HTTP request. + + :return: A tuple consisting of the session ID (a UUID string), and the session timeout date. + """ + # Reference date for session timeout + now = datetime.utcnow() + # Read the client's session ID, if any + session_id = str(request.get_cookie(_COOKIE_NAME, secret=__key)) + # If there is no active session, create a new session ID + if session_id is None: + session_id = str(uuid4()) + + # Check for session timeout + if session_id in __session_vars and __session_vars[session_id][0] < now: + end(session_id) + raise TimeoutError('Session timed out.') + # Update or initialize the session timeout + if session_id not in __session_vars: + __session_vars[session_id] = (now + timedelta(seconds=_SESSION_TIMEOUT)), dict() + else: + __session_vars[session_id] = \ + (now + timedelta(seconds=_SESSION_TIMEOUT), __session_vars[session_id][1]) + # Return the session ID and timeout + response.set_cookie(_COOKIE_NAME, session_id, secret=__key) + return session_id + + +def end(session_id: str) -> None: + """ + Destroy a session identified by the session ID. + + :param session_id: ID of the session to destroy. + """ + if session_id in __session_vars: + del __session_vars[session_id] + + +def put(session_id: str, key: str, value: Any) -> None: + if session_id in __session_vars: + __session_vars[session_id][1][key] = value + + +def get(session_id: str, key: str) -> Any: + if session_id in __session_vars and key in __session_vars[session_id][1]: + return __session_vars[session_id][1][key] + return None + + +def delete(session_id: str, key: str) -> None: + if session_id in __session_vars and key in __session_vars[session_id][1]: + del __session_vars[session_id][1][key] + + +def has(session_id: str, key: str) -> bool: + return session_id in __session_vars and key in __session_vars[session_id][1] diff --git a/matemat/webserver/template/__init__.py b/matemat/webserver/template/__init__.py new file mode 100644 index 0000000..919a08f --- /dev/null +++ b/matemat/webserver/template/__init__.py @@ -0,0 +1,2 @@ + +from .template import init, render diff --git a/matemat/webserver/template/template.py b/matemat/webserver/template/template.py new file mode 100644 index 0000000..94e6250 --- /dev/null +++ b/matemat/webserver/template/template.py @@ -0,0 +1,24 @@ +from typing import Any, Dict + +import os.path +import jinja2 + +from matemat import __version__ +from matemat.util.currency_format import format_chf + +__jinja_env: jinja2.Environment = None + + +def init(config: Dict[str, Any]) -> None: + global __jinja_env + __jinja_env = jinja2.Environment( + loader=jinja2.FileSystemLoader(os.path.abspath(config['templateroot'])), + autoescape=jinja2.select_autoescape(default=True), + ) + __jinja_env.filters['chf'] = format_chf + + +def render(name: str, **kwargs): + global __jinja_env + template: jinja2.Template = __jinja_env.get_template(name) + return template.render(__version__=__version__, **kwargs).encode('utf-8') diff --git a/matemat/webserver/test/test_pagelet_cron.py b/matemat/webserver/test/test_pagelet_cron.py index 4db29f8..162f91b 100644 --- a/matemat/webserver/test/test_pagelet_cron.py +++ b/matemat/webserver/test/test_pagelet_cron.py @@ -7,7 +7,7 @@ from threading import Lock, Thread, Timer from time import sleep import jinja2 -from matemat.webserver import MatematWebserver, pagelet_cron +from matemat.webserver.cron import cron lock: Lock = Lock() @@ -15,7 +15,7 @@ cron1called: int = 0 cron2called: int = 0 -@pagelet_cron(seconds=4) +@cron(seconds=4) def cron1(config: Dict[str, str], jinja_env: jinja2.Environment, logger: logging.Logger) -> None: @@ -24,7 +24,7 @@ def cron1(config: Dict[str, str], cron1called += 1 -@pagelet_cron(seconds=3) +@cron(seconds=3) def cron2(config: Dict[str, str], jinja_env: jinja2.Environment, logger: logging.Logger) -> None: diff --git a/matemat/webserver/test/test_parse_request.py b/matemat/webserver/test/test_parse_request.py deleted file mode 100644 index 8f3d35b..0000000 --- a/matemat/webserver/test/test_parse_request.py +++ /dev/null @@ -1,437 +0,0 @@ - -import unittest - -from matemat.webserver.util import parse_args - - -class TestParseRequest(unittest.TestCase): - - def test_parse_get_root(self): - """ - Test that the simple root path is parsed correctly ('/' path, no args). - """ - path, args = parse_args('/') - self.assertEqual('/', path) - self.assertEqual(0, len(args)) - - def test_parse_get_no_args(self): - """ - Test that a GET request without arguments is parsed correctly. - """ - path, args = parse_args('/index.html') - self.assertEqual('/index.html', path) - self.assertEqual(0, len(args)) - - def test_parse_get_root_getargs(self): - """ - Test that a GET request for '/' with scalar arguments is parsed correctly. - """ - path, args = parse_args('/?foo=42&bar=1337&baz=Hello,%20World!') - self.assertEqual('/', path) - self.assertEqual(3, len(args)) - self.assertIn('foo', args) - self.assertIn('bar', args) - self.assertIn('baz', args) - self.assertTrue(args['foo'].is_scalar) - self.assertTrue(args['bar'].is_scalar) - self.assertTrue(args['baz'].is_scalar) - self.assertEqual('text/plain', args['foo'].get_content_type()) - self.assertEqual('text/plain', args['bar'].get_content_type()) - self.assertEqual('text/plain', args['baz'].get_content_type()) - self.assertEqual('42', args['foo'].get_str()) - self.assertEqual('1337', args['bar'].get_str()) - self.assertEqual('Hello, World!', args['baz'].get_str()) - - def test_parse_get_getargs(self): - """ - Test that a GET request for an arbitrary path with scalar arguments is parsed correctly. - """ - path, args = parse_args('/abc/def?foo=42&bar=1337&baz=Hello,%20World!') - self.assertEqual('/abc/def', path) - self.assertEqual(3, len(args)) - self.assertIn('foo', args) - self.assertIn('bar', args) - self.assertIn('baz', args) - self.assertTrue(args['foo'].is_scalar) - self.assertTrue(args['bar'].is_scalar) - self.assertTrue(args['baz'].is_scalar) - self.assertEqual('text/plain', args['foo'].get_content_type()) - self.assertEqual('text/plain', args['bar'].get_content_type()) - self.assertEqual('text/plain', args['baz'].get_content_type()) - self.assertEqual('42', args['foo'].get_str()) - self.assertEqual('1337', args['bar'].get_str()) - self.assertEqual('Hello, World!', args['baz'].get_str()) - - def test_parse_get_getarray(self): - """ - Test that a GET request with mixed scalar and array arguments is parsed correctly. - """ - path, args = parse_args('/abc/def?foo=42&foo=1337&baz=Hello,%20World!') - self.assertEqual('/abc/def', path) - self.assertEqual(2, len(args)) - self.assertIn('foo', args) - self.assertIn('baz', args) - self.assertTrue(args['foo'].is_array) - self.assertTrue(args['baz'].is_scalar) - self.assertEqual(2, len(args['foo'])) - self.assertEqual('42', args['foo'].get_str(0)) - self.assertEqual('1337', args['foo'].get_str(1)) - - def test_parse_get_zero_arg(self): - """ - Test that a GET request with an empty argument is parsed correctly. - """ - path, args = parse_args('/abc/def?foo=&bar=42') - self.assertEqual(2, len(args)) - self.assertIn('foo', args) - self.assertIn('bar', args) - self.assertTrue(args['foo'].is_scalar) - self.assertTrue(args['bar'].is_scalar) - self.assertEqual(1, len(args['foo'])) - self.assertEqual('', args['foo'].get_str()) - self.assertEqual('42', args['bar'].get_str()) - - def test_parse_get_urlencoded_encoding_fail(self): - """ - Test that a GET request with non-decodable escape sequences fails. - """ - with self.assertRaises(ValueError): - parse_args('/?foo=42&bar=%80&baz=Hello,%20World!') - - def test_parse_post_urlencoded(self): - """ - Test that a urlencoded POST request with scalar arguments is parsed correctly. - """ - path, args = parse_args('/', - postbody=b'foo=42&bar=1337&baz=Hello,%20World!', - enctype='application/x-www-form-urlencoded') - self.assertEqual('/', path) - self.assertEqual(3, len(args)) - self.assertIn('foo', args) - self.assertIn('bar', args) - self.assertIn('baz', args) - self.assertTrue(args['foo'].is_scalar) - self.assertTrue(args['bar'].is_scalar) - self.assertTrue(args['baz'].is_scalar) - self.assertEqual('text/plain', args['foo'].get_content_type()) - self.assertEqual('text/plain', args['bar'].get_content_type()) - self.assertEqual('text/plain', args['baz'].get_content_type()) - self.assertEqual('42', args['foo'].get_str()) - self.assertEqual('1337', args['bar'].get_str()) - self.assertEqual('Hello, World!', args['baz'].get_str()) - - def test_parse_post_urlencoded_array(self): - """ - Test that a urlencoded POST request with mixed scalar and array arguments is parsed correctly. - """ - path, args = parse_args('/', - postbody=b'foo=42&foo=1337&baz=Hello,%20World!', - enctype='application/x-www-form-urlencoded') - self.assertEqual('/', path) - self.assertEqual(2, len(args)) - self.assertIn('foo', args) - self.assertIn('baz', args) - self.assertTrue(args['foo'].is_array) - self.assertTrue(args['baz'].is_scalar) - self.assertEqual(2, len(args['foo'])) - self.assertEqual('42', args['foo'].get_str(0)) - self.assertEqual('1337', args['foo'].get_str(1)) - - def test_parse_post_urlencoded_zero_arg(self): - """ - Test that a urlencoded POST request with an empty argument is parsed correctly. - """ - path, args = parse_args('/abc/def', postbody=b'foo=&bar=42', enctype='application/x-www-form-urlencoded') - self.assertEqual(2, len(args)) - self.assertIn('foo', args) - self.assertIn('bar', args) - self.assertTrue(args['foo'].is_scalar) - self.assertTrue(args['bar'].is_scalar) - self.assertEqual(1, len(args['foo'])) - self.assertEqual('', args['foo'].get_str()) - self.assertEqual('42', args['bar'].get_str()) - - def test_parse_post_urlencoded_encoding_fail(self): - """ - Test that a urlencoded POST request with a non-decodable escape sequence fails. - """ - with self.assertRaises(ValueError): - parse_args('/', - postbody=b'foo=42&bar=%80&baz=Hello,%20World!', - enctype='application/x-www-form-urlencoded') - - def test_parse_post_multipart_no_args(self): - """ - Test that a multipart POST request with no arguments is parsed correctly. - """ - path, args = parse_args('/', - postbody=b'--testBoundary1337--\r\n', - enctype='multipart/form-data; boundary=testBoundary1337') - self.assertEqual('/', path) - self.assertEqual(0, len(args)) - - def test_parse_post_multipart(self): - """ - Test that a multipart POST request with scalar arguments is parsed correctly. - """ - path, args = parse_args('/', - postbody=b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="foo"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'42\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="bar"; filename="bar.bin"\r\n' - b'Content-Type: application/octet-stream\r\n\r\n' - b'1337\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="baz"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'Hello, World!\r\n' - b'--testBoundary1337--\r\n', - enctype='multipart/form-data; boundary=testBoundary1337') - self.assertEqual('/', path) - self.assertEqual(3, len(args)) - self.assertIn('foo', args) - self.assertIn('bar', args) - self.assertIn('baz', args) - self.assertTrue(args['foo'].is_scalar) - self.assertTrue(args['bar'].is_scalar) - self.assertTrue(args['baz'].is_scalar) - self.assertEqual('text/plain', args['foo'].get_content_type()) - self.assertEqual('application/octet-stream', args['bar'].get_content_type()) - self.assertEqual('text/plain', args['baz'].get_content_type()) - self.assertEqual('42', args['foo'].get_str()) - self.assertEqual(b'1337', args['bar'].get_bytes()) - self.assertEqual('Hello, World!', args['baz'].get_str()) - - def test_parse_post_multipart_names(self): - """ - Test that multipart names work both with and without quotation marks - """ - path, args = parse_args('/', - postbody=b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="foo"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'42\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name=bar\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'Hello, World!\r\n' - b'--testBoundary1337--\r\n', - enctype='multipart/form-data; boundary=testBoundary1337') - self.assertEqual('/', path) - self.assertEqual(2, len(args)) - self.assertIn('foo', args) - self.assertIn('bar', args) - self.assertTrue(args['foo'].is_scalar) - self.assertTrue(args['bar'].is_scalar) - self.assertEqual('text/plain', args['foo'].get_content_type()) - self.assertEqual('text/plain', args['bar'].get_content_type()) - self.assertEqual('42', args['foo'].get_str()) - self.assertEqual('Hello, World!', args['bar'].get_str()) - - def test_parse_post_multipart_zero_arg(self): - """ - Test that a multipart POST request with an empty argument is parsed correctly. - """ - path, args = parse_args('/abc/def', - postbody=b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="foo"\r\n' - b'Content-Type: text/plain\r\n\r\n\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="bar"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'42\r\n' - b'--testBoundary1337--\r\n', - enctype='multipart/form-data; boundary=testBoundary1337') - self.assertEqual(2, len(args)) - self.assertIn('foo', args) - self.assertIn('bar', args) - self.assertTrue(args['foo'].is_scalar) - self.assertTrue(args['bar'].is_scalar) - self.assertEqual(1, len(args['foo'])) - self.assertEqual('', args['foo'].get_str()) - self.assertEqual('42', args['bar'].get_str()) - - def test_parse_post_multipart_no_contenttype(self): - """ - Test that the Content-Type is set to 'application/octet-stream' if it is absent from the multipart header. - """ - path, args = parse_args('/', - postbody=b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="foo"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'42\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="bar"; filename="bar.bin"\r\n' - b'Content-Type: application/octet-stream\r\n\r\n' - b'1337\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="baz"\r\n\r\n' - b'Hello, World!\r\n' - b'--testBoundary1337--\r\n', - enctype='multipart/form-data; boundary=testBoundary1337') - self.assertEqual('/', path) - self.assertEqual(3, len(args)) - self.assertIn('foo', args) - self.assertIn('bar', args) - self.assertIn('baz', args) - self.assertTrue(args['foo'].is_scalar) - self.assertTrue(args['bar'].is_scalar) - self.assertTrue(args['baz'].is_scalar) - self.assertEqual('application/octet-stream', args['baz'].get_content_type()) - self.assertEqual('Hello, World!', args['baz'].get_str()) - - def test_parse_post_multipart_broken_boundaries(self): - """ - Test that multiple cases with broken multipart boundaries fail. - """ - with self.assertRaises(ValueError): - # Boundary not defined in Content-Type - parse_args('/', - postbody=b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="foo"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'42\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="bar"; filename="bar.bin"\r\n' - b'Content-Type: application/octet-stream\r\n\r\n' - b'1337\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="baz"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'Hello, World!\r\n' - b'--testBoundary1337--\r\n', - enctype='multipart/form-data') - with self.assertRaises(ValueError): - # Corrupted "--" head at first boundary - parse_args('/', - postbody=b'-+testBoundary1337\r\n' - b'Content-Disposition: form-data; name="foo"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'42\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="bar"; filename="bar.bin"\r\n' - b'Content-Type: application/octet-stream\r\n\r\n' - b'1337\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="baz"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'Hello, World!\r\n' - b'--testBoundary1337--\r\n', - enctype='multipart/form-data; boundary=testBoundary1337') - with self.assertRaises(ValueError): - # Missing "--" tail at end boundary - parse_args('/', - postbody=b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="foo"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'42\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="bar"; filename="bar.bin"\r\n' - b'Content-Type: application/octet-stream\r\n\r\n' - b'1337\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="baz"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'Hello, World!\r\n' - b'--testBoundary1337\r\n', - enctype='multipart/form-data; boundary=testBoundary1337') - with self.assertRaises(ValueError): - # Missing Content-Disposition header in one part - parse_args('/', - postbody=b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="foo"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'42\r\n' - b'--testBoundary1337\r\n' - b'Content-Type: application/octet-stream\r\n\r\n' - b'1337\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="baz"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'Hello, World!\r\n' - b'--testBoundary1337--\r\n', - enctype='multipart/form-data; boundary=testBoundary1337') - with self.assertRaises(ValueError): - # Missing form-data name argument - parse_args('/', - postbody=b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="foo"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'42\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; filename="bar.bin"\r\n' - b'Content-Type: application/octet-stream\r\n\r\n' - b'1337\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="baz"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'Hello, World!\r\n' - b'--testBoundary1337--\r\n', - enctype='multipart/form-data; boundary=testBoundary1337') - with self.assertRaises(ValueError): - # Unknown Content-Disposition - parse_args('/', - postbody=b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="foo"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'42\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: attachment; name="bar"; filename="bar.bin"\r\n' - b'Content-Type: application/octet-stream\r\n\r\n' - b'1337\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="baz"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'Hello, World!\r\n' - b'--testBoundary1337--\r\n', - enctype='multipart/form-data; boundary=testBoundary1337') - - def test_get_post_precedence_urlencoded(self): - """ - Test the precedence of urlencoded POST arguments over GET arguments. - """ - path, args = parse_args('/foo?foo=thisshouldnotbethere&bar=isurvived', - postbody=b'foo=42&foo=1337&baz=Hello,%20World!', - enctype='application/x-www-form-urlencoded') - self.assertIn('foo', args) - self.assertIn('bar', args) - self.assertIn('baz', args) - self.assertEqual(2, len(args['foo'])) - self.assertEqual(1, len(args['bar'])) - self.assertEqual(1, len(args['baz'])) - self.assertEqual('42', args['foo'].get_str(0)) - self.assertEqual('1337', args['foo'].get_str(1)) - self.assertEqual('isurvived', args['bar'].get_str()) - self.assertEqual('Hello, World!', args['baz'].get_str()) - - def test_get_post_precedence_multipart(self): - """ - Test the precedence of multipart POST arguments over GET arguments. - """ - path, args = parse_args('/foo?foo=thisshouldnotbethere&bar=isurvived', - postbody=b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="foo"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'42\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="foo"; filename="bar.bin"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'1337\r\n' - b'--testBoundary1337\r\n' - b'Content-Disposition: form-data; name="baz"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'Hello, World!\r\n' - b'--testBoundary1337--\r\n', - enctype='multipart/form-data; boundary=testBoundary1337') - self.assertIn('foo', args) - self.assertIn('bar', args) - self.assertIn('baz', args) - self.assertEqual(2, len(args['foo'])) - self.assertEqual(1, len(args['bar'])) - self.assertEqual(1, len(args['baz'])) - self.assertEqual('42', args['foo'].get_str(0)) - self.assertEqual('1337', args['foo'].get_str(1)) - self.assertEqual('isurvived', args['bar'].get_str()) - self.assertEqual('Hello, World!', args['baz'].get_str()) diff --git a/matemat/webserver/test/test_post.py b/matemat/webserver/test/test_post.py deleted file mode 100644 index a006221..0000000 --- a/matemat/webserver/test/test_post.py +++ /dev/null @@ -1,225 +0,0 @@ - -from typing import Any, Dict, List - -from matemat.webserver import HttpHandler, RequestArguments -from matemat.webserver.test.abstract_httpd_test import AbstractHttpdTest, test_pagelet - -import codecs - - -@test_pagelet('/just/testing/post') -def post_test_pagelet(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - pagelet_variables: Dict[str, str]): - """ - Test pagelet that simply prints the parsed arguments as response body. - """ - headers['Content-Type'] = 'text/plain' - dump: str = '' - for ra in args: - for a in ra: - if a.get_content_type().startswith('text/'): - dump += f'{a.name}: {a.get_str()}\n' - else: - dump += f'{a.name}: {codecs.encode(a.get_bytes(), "hex").decode("utf-8")}\n' - return dump - - -class TestPost(AbstractHttpdTest): - """ - Test cases for the content serving of the web server. - """ - - def test_post_urlenc_get_only_args(self): - """ - Test a POST request that only contains GET arguments. - """ - # Send POST request - self.client_sock.set_request(b'POST /just/testing/post?foo=bar&test=1 HTTP/1.1\r\n' - b'Content-Length: 0\r\n' - b'Content-Type: application/x-www-form-urlencoded\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Parse response body - lines: List[bytes] = packet.body.split(b'\n')[:-1] - kv: Dict[str, str] = dict() - for l in lines: - k, v = l.decode('utf-8').split(':', 1) - kv[k.strip()] = v.strip() if ',' not in v else v.strip().split(',') - - # Make sure the arguments were properly parsed - self.assertEqual('bar', kv['foo']) - self.assertEqual('1', kv['test']) - - def test_post_urlenc_post_only_args(self): - """ - Test a POST request that only contains POST arguments (urlencoded). - """ - # Send POST request - self.client_sock.set_request(b'POST /just/testing/post HTTP/1.1\r\n' - b'Content-Type: application/x-www-form-urlencoded\r\n' - b'Content-Length: 14\r\n\r\n' - b'foo=bar&test=1\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Parse response body - lines: List[bytes] = packet.body.split(b'\n')[:-1] - kv: Dict[str, str] = dict() - for l in lines: - k, v = l.decode('utf-8').split(':', 1) - kv[k.strip()] = v.strip() if ',' not in v else v.strip().split(',') - - # Make sure the arguments were properly parsed - self.assertEqual('bar', kv['foo']) - self.assertEqual('1', kv['test']) - - def test_post_urlenc_mixed_args(self): - """ - Test that mixed POST and GET args are properly parsed, and that POST takes precedence over GET. - """ - # Send POST request - self.client_sock.set_request(b'POST /just/testing/post?gettest=1&foo=baz HTTP/1.1\r\n' - b'Content-Type: application/x-www-form-urlencoded\r\n' - b'Content-Length: 18\r\n\r\n' - b'foo=bar&posttest=2\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Parse response body - lines: List[bytes] = packet.body.split(b'\n')[:-1] - kv: Dict[str, str] = dict() - for l in lines: - k, v = l.decode('utf-8').split(':', 1) - kv[k.strip()] = v.strip() if ',' not in v else v.strip().split(',') - - # Make sure the arguments were properly parsed - self.assertEqual('bar', kv['foo']) - self.assertEqual('1', kv['gettest']) - self.assertEqual('2', kv['posttest']) - - def test_post_urlenc_get_array(self): - """ - Test a POST request that contains GET array arguments. - """ - # Send POST request - self.client_sock.set_request(b'POST /just/testing/post?foo=bar&test=1&foo=baz HTTP/1.1\r\n' - b'Content-Length: 0\r\n' - b'Content-Type: application/x-www-form-urlencoded\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Parse response body - lines: List[bytes] = packet.body.split(b'\n')[:-1] - kv: Dict[str, str] = dict() - for l in lines: - k, v = l.decode('utf-8').split(':', 1) - k = k.strip() - v = v.strip() - if k in kv: - kv[k] += f',{v}' - else: - kv[k] = v - # Make sure the arguments were properly parsed - self.assertEqual('bar,baz', kv['foo']) - self.assertEqual('1', kv['test']) - - def test_post_urlenc_post_array(self): - """ - Test a POST request that contains POST array arguments. - """ - # Send POST request - self.client_sock.set_request(b'POST /just/testing/post HTTP/1.1\r\n' - b'Content-Length: 22\r\n' - b'Content-Type: application/x-www-form-urlencoded\r\n\r\n' - b'foo=bar&test=1&foo=baz\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Parse response body - lines: List[bytes] = packet.body.split(b'\n')[:-1] - kv: Dict[str, str] = dict() - for l in lines: - k, v = l.decode('utf-8').split(':', 1) - k = k.strip() - v = v.strip() - if k in kv: - kv[k] += f',{v}' - else: - kv[k] = v - # Make sure the arguments were properly parsed - self.assertEqual('bar,baz', kv['foo']) - self.assertEqual('1', kv['test']) - - def test_post_urlenc_mixed_array(self): - """ - Test a POST request that contains both GET and POST array arguments. - """ - # Send POST request - self.client_sock.set_request(b'POST /just/testing/post?foo=getbar&gettest=1&gettest=42&foo=getbaz HTTP/1.1\r\n' - b'Content-Length: 45\r\n' - b'Content-Type: application/x-www-form-urlencoded\r\n\r\n' - b'foo=postbar&posttest=1&posttest=2&foo=postbaz\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Parse response body - lines: List[bytes] = packet.body.split(b'\n')[:-1] - kv: Dict[str, str] = dict() - for l in lines: - k, v = l.decode('utf-8').split(':', 1) - k = k.strip() - v = v.strip() - if k in kv: - kv[k] += f',{v}' - else: - kv[k] = v - # Make sure the arguments were properly parsed - self.assertEqual('postbar,postbaz', kv['foo']) - self.assertEqual('1,42', kv['gettest']) - self.assertEqual('1,2', kv['posttest']) - - def test_post_no_body(self): - """ - Test a POST request that contains no headers or body. - """ - # Send POST request - self.client_sock.set_request(b'POST /just/testing/post?foo=bar HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - # Make sure a 400 Bad Request is returned - self.assertEqual(400, packet.statuscode) - - def test_post_multipart_post_only(self): - """ - Test a POST request with a miutipart/form-data body. - """ - # Send POST request - formdata = (b'------testboundary\r\n' - b'Content-Disposition: form-data; name="foo"\r\n' - b'Content-Type: text/plain\r\n\r\n' - b'Hello, World!\r\n' - b'------testboundary\r\n' - b'Content-Disposition: form-data; name="bar"; filename="foo.bar"\r\n' - b'Content-Type: application/octet-stream\r\n\r\n' - b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x80\x0b\x0c\x73\x0e\x0f\r\n' - b'------testboundary--\r\n') - - self.client_sock.set_request(f'POST /just/testing/post HTTP/1.1\r\n' - f'Content-Type: multipart/form-data; boundary=----testboundary\r\n' - f'Content-Length: {len(formdata)}\r\n\r\n'.encode('utf-8') + formdata) - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - lines: List[bytes] = packet.body.split(b'\n')[:-1] - kv: Dict[str, Any] = dict() - for l in lines: - k, v = l.split(b':', 1) - kv[k.decode('utf-8').strip()] = v.strip() - self.assertIn('foo', kv) - self.assertIn('bar', kv) - self.assertEqual(kv['foo'], b'Hello, World!') - self.assertEqual(kv['bar'], b'00010203040506070809800b0c730e0f') diff --git a/matemat/webserver/test/test_requestargs.py b/matemat/webserver/test/test_requestargs.py deleted file mode 100644 index 3e093a2..0000000 --- a/matemat/webserver/test/test_requestargs.py +++ /dev/null @@ -1,529 +0,0 @@ - -from typing import Dict, List, Set, Tuple - -import unittest -import urllib.parse - -from matemat.webserver import RequestArgument, RequestArguments -# noinspection PyProtectedMember -from matemat.webserver.requestargs import _View - - -class TestRequestArguments(unittest.TestCase): - """ - Test cases for the RequestArgument class. - """ - - def test_create_default(self): - """ - Test creation of an empty RequestArgument - """ - ra = RequestArgument('foo') - # Name must be set to 1st argument - self.assertEqual('foo', ra.name) - # Must be a 0-length array - self.assertEqual(0, len(ra)) - self.assertFalse(ra.is_scalar) - self.assertTrue(ra.is_array) - # Must not be a view - self.assertFalse(ra.is_view) - - def test_create_str_scalar(self): - """ - Test creation of a scalar RequestArgument with string value. - """ - ra = RequestArgument('foo', ('text/plain', 'bar')) - # Name must be set to 1st argument - self.assertEqual('foo', ra.name) - # Must be a scalar, length 1 - self.assertEqual(1, len(ra)) - self.assertTrue(ra.is_scalar) - self.assertFalse(ra.is_array) - # Scalar value must be representable both as str and bytes - self.assertEqual('bar', ra.get_str()) - self.assertEqual(b'bar', ra.get_bytes()) - # Content-Type must be set correctly - self.assertEqual('text/plain', ra.get_content_type()) - # Using 0 indices must yield the same results - self.assertEqual('bar', ra.get_str(0)) - self.assertEqual(b'bar', ra.get_bytes(0)) - self.assertEqual('text/plain', ra.get_content_type(0)) - # Using other indices must result in an error - with self.assertRaises(IndexError): - ra.get_str(1) - with self.assertRaises(IndexError): - ra.get_bytes(1) - with self.assertRaises(IndexError): - ra.get_content_type(1) - # Must not be a view - self.assertFalse(ra.is_view) - - def test_create_str_scalar_array(self): - """ - Test creation of a scalar RequestArgument with string value, passing an array instead of a single tuple. - """ - ra = RequestArgument('foo', [('text/plain', 'bar')]) - # Name must be set to 1st argument - self.assertEqual('foo', ra.name) - # Must be a scalar, length 1 - self.assertEqual(1, len(ra)) - self.assertTrue(ra.is_scalar) - self.assertFalse(ra.is_array) - # Scalar value must be representable both as str and bytes - self.assertEqual('bar', ra.get_str()) - self.assertEqual(b'bar', ra.get_bytes()) - # Content-Type must be set correctly - self.assertEqual('text/plain', ra.get_content_type()) - # Using 0 indices must yield the same results - self.assertEqual('bar', ra.get_str(0)) - self.assertEqual(b'bar', ra.get_bytes(0)) - self.assertEqual('text/plain', ra.get_content_type(0)) - # Using other indices must result in an error - with self.assertRaises(IndexError): - ra.get_str(1) - with self.assertRaises(IndexError): - ra.get_bytes(1) - with self.assertRaises(IndexError): - ra.get_content_type(1) - # Must not be a view - self.assertFalse(ra.is_view) - - def test_create_bytes_scalar(self): - """ - Test creation of a scalar RequestArgument with bytes value. - """ - ra = RequestArgument('foo', ('application/octet-stream', b'\x00\x80\xff\xfe')) - # Name must be set to 1st argument - self.assertEqual('foo', ra.name) - # Must be a scalar, length 1 - self.assertEqual(1, len(ra)) - self.assertTrue(ra.is_scalar) - self.assertFalse(ra.is_array) - # Conversion to UTF-8 string must fail; bytes representation must work - with self.assertRaises(UnicodeDecodeError): - ra.get_str() - self.assertEqual(b'\x00\x80\xff\xfe', ra.get_bytes()) - # Content-Type must be set correctly - self.assertEqual('application/octet-stream', ra.get_content_type()) - # Using 0 indices must yield the same results - with self.assertRaises(UnicodeDecodeError): - ra.get_str(0) - self.assertEqual(b'\x00\x80\xff\xfe', ra.get_bytes(0)) - self.assertEqual('application/octet-stream', ra.get_content_type(0)) - # Using other indices must result in an error - with self.assertRaises(IndexError): - ra.get_str(1) - with self.assertRaises(IndexError): - ra.get_bytes(1) - with self.assertRaises(IndexError): - ra.get_content_type(1) - # Must not be a view - self.assertFalse(ra.is_view) - - def test_create_array(self): - """ - Test creation of an array RequestArgument with mixed str and bytes initial value. - """ - ra = RequestArgument('foo', [ - ('text/plain', 'bar'), - ('application/octet-stream', b'\x00\x80\xff\xfe') - ]) - # Name must be set to 1st argument - self.assertEqual('foo', ra.name) - # Must be an array, length 2 - self.assertEqual(2, len(ra)) - self.assertFalse(ra.is_scalar) - self.assertTrue(ra.is_array) - # Retrieving values without an index must yield the first element - self.assertEqual('bar', ra.get_str()) - self.assertEqual(b'bar', ra.get_bytes()) - self.assertEqual('text/plain', ra.get_content_type()) - # The first value must be representable both as str and bytes, and have ctype text/plain - self.assertEqual('bar', ra.get_str(0)) - self.assertEqual(b'bar', ra.get_bytes(0)) - self.assertEqual('text/plain', ra.get_content_type(0)) - # Conversion of the second value to UTF-8 string must fail; bytes representation must work - with self.assertRaises(UnicodeDecodeError): - ra.get_str(1) - self.assertEqual(b'\x00\x80\xff\xfe', ra.get_bytes(1)) - # The second value's ctype must be correct - self.assertEqual('application/octet-stream', ra.get_content_type(1)) - # Must not be a view - self.assertFalse(ra.is_view) - - def test_append_empty_str(self): - """ - Test appending a str value to an empty RequestArgument. - """ - # Initialize the empty RequestArgument - ra = RequestArgument('foo') - self.assertEqual(0, len(ra)) - self.assertFalse(ra.is_scalar) - - # Append a string value - ra.append('text/plain', 'bar') - # New length must be 1, empty array must be converted to scalar - self.assertEqual(1, len(ra)) - self.assertTrue(ra.is_scalar) - # Retrieval of the new value must work both in str and bytes representation - self.assertEqual('bar', ra.get_str()) - self.assertEqual(b'bar', ra.get_bytes()) - # Content type of the new value must be correct - self.assertEqual('text/plain', ra.get_content_type()) - # Must not be a view - self.assertFalse(ra.is_view) - - def test_append_empty_bytes(self): - """ - Test appending a bytes value to an empty RequestArgument. - """ - # Initialize the empty RequestArgument - ra = RequestArgument('foo') - self.assertEqual(0, len(ra)) - self.assertFalse(ra.is_scalar) - - # Append a bytes value - ra.append('application/octet-stream', b'\x00\x80\xff\xfe') - # New length must be 1, empty array must be converted to scalar - self.assertEqual(1, len(ra)) - self.assertTrue(ra.is_scalar) - # Conversion of the new value to UTF-8 string must fail; bytes representation must work - with self.assertRaises(UnicodeDecodeError): - ra.get_str() - self.assertEqual(b'\x00\x80\xff\xfe', ra.get_bytes()) - # Content type of the new value must be correct - self.assertEqual('application/octet-stream', ra.get_content_type()) - # Must not be a view - self.assertFalse(ra.is_view) - - def test_append_multiple(self): - """ - Test appending multiple values to an empty RequestArgument. - """ - # Initialize the empty RequestArgument - ra = RequestArgument('foo') - self.assertEqual(0, len(ra)) - self.assertFalse(ra.is_scalar) - - # Append a first value - ra.append('text/plain', 'bar') - # New length must be 1, empty array must be converted to scalar - self.assertEqual(1, len(ra)) - self.assertTrue(ra.is_scalar) - self.assertEqual(b'bar', ra.get_bytes()) - - # Append a second value - ra.append('application/octet-stream', b'\x00\x80\xff\xfe') - # New length must be 2, scalar must be converted to array - self.assertEqual(2, len(ra)) - self.assertFalse(ra.is_scalar) - self.assertEqual(b'bar', ra.get_bytes(0)) - self.assertEqual(b'\x00\x80\xff\xfe', ra.get_bytes(1)) - - # Append a third value - ra.append('text/plain', 'Hello, World!') - # New length must be 3, array must remain array - self.assertEqual(3, len(ra)) - self.assertFalse(ra.is_scalar) - self.assertEqual(b'bar', ra.get_bytes(0)) - self.assertEqual(b'\x00\x80\xff\xfe', ra.get_bytes(1)) - self.assertEqual(b'Hello, World!', ra.get_bytes(2)) - - def test_clear_empty(self): - """ - Test clearing an empty RequestArgument. - """ - # Initialize the empty RequestArgument - ra = RequestArgument('foo') - self.assertEqual(0, len(ra)) - self.assertFalse(ra.is_scalar) - ra.clear() - # Clearing an empty RequestArgument shouldn't have any effect - self.assertEqual('foo', ra.name) - self.assertEqual(0, len(ra)) - self.assertFalse(ra.is_scalar) - - def test_clear_scalar(self): - """ - Test clearing a scalar RequestArgument. - """ - # Initialize the scalar RequestArgument - ra = RequestArgument('foo', ('text/plain', 'bar')) - self.assertEqual(1, len(ra)) - self.assertTrue(ra.is_scalar) - ra.clear() - # Clearing a scalar RequestArgument should reduce its size to 0 - self.assertEqual('foo', ra.name) - self.assertEqual(0, len(ra)) - self.assertFalse(ra.is_scalar) - with self.assertRaises(IndexError): - ra.get_str() - - def test_clear_array(self): - """ - Test clearing an array RequestArgument. - """ - # Initialize the array RequestArgument - ra = RequestArgument('foo', [ - ('text/plain', 'bar'), - ('application/octet-stream', b'\x00\x80\xff\xfe'), - ('text/plain', 'baz'), - ]) - self.assertEqual(3, len(ra)) - self.assertFalse(ra.is_scalar) - ra.clear() - # Clearing an array RequestArgument should reduce its size to 0 - self.assertEqual('foo', ra.name) - self.assertEqual(0, len(ra)) - self.assertFalse(ra.is_scalar) - with self.assertRaises(IndexError): - ra.get_str() - - def test_iterate_empty(self): - """ - Test iterating an empty RequestArgument. - """ - ra = RequestArgument('foo') - self.assertEqual(0, len(ra)) - # No value must be yielded from iterating an empty instance - for _ in ra: - self.fail() - - def test_iterate_scalar(self): - """ - Test iterating a scalar RequestArgument. - """ - ra = RequestArgument('foo', ('text/plain', 'bar')) - self.assertTrue(ra.is_scalar) - # Counter for the number of iterations - count: int = 0 - for it in ra: - # Make sure the yielded value is a scalar view and has the same name as the original instance - self.assertIsInstance(it, _View) - self.assertTrue(it.is_view) - self.assertEqual('foo', it.name) - self.assertTrue(it.is_scalar) - count += 1 - # Only one value must be yielded from iterating a scalar instance - self.assertEqual(1, count) - - def test_iterate_array(self): - """ - Test iterating an array RequestArgument. - """ - ra = RequestArgument('foo', [('text/plain', 'bar'), ('abc', b'def'), ('xyz', '1337')]) - self.assertFalse(ra.is_scalar) - # Container to put the iterated ctypes into - items: List[str] = list() - for it in ra: - # Make sure the yielded values are scalar views and have the same name as the original instance - self.assertIsInstance(it, _View) - self.assertTrue(it.is_view) - self.assertTrue(it.is_scalar) - # Collect the value's ctype - items.append(it.get_content_type()) - # Compare collected ctypes with expected result - self.assertEqual(['text/plain', 'abc', 'xyz'], items) - - def test_slice(self): - """ - Test slicing an array RequestArgument. - """ - ra = RequestArgument('foo', [('a', 'b'), ('c', 'd'), ('e', 'f'), ('g', 'h'), ('i', 'j'), ('k', 'l')]) - # Create the sliced view - sliced = ra[1:4:2] - # Make sure the sliced value is a view - self.assertIsInstance(sliced, _View) - self.assertTrue(sliced.is_view) - # Make sure the slice has the same name - self.assertEqual('foo', sliced.name) - # Make sure the slice has the expected shape (array of the 2nd and 4th scalar in the original) - self.assertTrue(sliced.is_array) - self.assertEqual(2, len(sliced)) - self.assertEqual('d', sliced.get_str(0)) - self.assertEqual('h', sliced.get_str(1)) - - def test_iterate_sliced(self): - """ - Test iterating a sliced array RequestArgument. - """ - ra = RequestArgument('foo', [('a', 'b'), ('c', 'd'), ('e', 'f'), ('g', 'h'), ('i', 'j'), ('k', 'l')]) - # Container to put the iterated ctypes into - items: List[str] = list() - # Iterate the sliced view - for it in ra[1:4:2]: - # Make sure the yielded values are scalar views and have the same name as the original instance - self.assertIsInstance(it, _View) - self.assertTrue(it.is_view) - self.assertEqual('foo', it.name) - self.assertTrue(it.is_scalar) - items.append(it.get_content_type()) - # Make sure the expected values are collected (array of the 2nd and 4th scalar in the original) - self.assertEqual(['c', 'g'], items) - - def test_index_scalar(self): - """ - Test indexing of a scalar RequestArgument. - """ - ra = RequestArgument('foo', ('bar', 'baz')) - # Index the scalar RequestArgument instance, obtaining an immutable view - it = ra[0] - # Make sure the value is a scalar view with the same properties as the original instance - self.assertIsInstance(it, _View) - self.assertTrue(it.is_scalar) - self.assertEqual('foo', it.name) - self.assertEqual('bar', it.get_content_type()) - self.assertEqual('baz', it.get_str()) - # Make sure other indices don't work - with self.assertRaises(IndexError): - _ = ra[1] - - def test_index_array(self): - """ - Test indexing of an array RequestArgument. - """ - ra = RequestArgument('foo', [('a', 'b'), ('c', 'd')]) - # Index the array RequestArgument instance, obtaining an immutable view - it = ra[1] - # Make sure the value is a scalar view with the same properties as the value in the original instance - self.assertIsInstance(it, _View) - self.assertEqual('foo', it.name) - self.assertEqual('c', it.get_content_type()) - self.assertEqual('d', it.get_str()) - - def test_view_immutable(self): - """ - Test immutability of views. - """ - ra = RequestArgument('foo', ('bar', 'baz')) - # Index the scalar RequestArgument instance, obtaining an immutable view - it = ra[0] - # Make sure the returned value is a view - self.assertIsInstance(it, _View) - # Make sure the returned value is immutable - with self.assertRaises(TypeError): - it.append('foo', 'bar') - with self.assertRaises(TypeError): - it.clear() - - def test_str_shorthand(self): - """ - Test the shorthand for get_str(0). - """ - ra = RequestArgument('foo', ('bar', 'baz')) - self.assertEqual('baz', str(ra)) - - def test_bytes_shorthand(self): - """ - Test the shorthand for get_bytes(0). - """ - ra = RequestArgument('foo', ('bar', b'\x00\x80\xff\xfe')) - self.assertEqual(b'\x00\x80\xff\xfe', bytes(ra)) - - # noinspection PyTypeChecker - def test_insert_garbage(self): - """ - Test proper handling with non-int indices and non-str/non-bytes data - :return: - """ - ra = RequestArgument('foo', 42) - with self.assertRaises(TypeError): - str(ra) - ra = RequestArgument('foo', (None, 42)) - with self.assertRaises(TypeError): - str(ra) - with self.assertRaises(TypeError): - bytes(ra) - with self.assertRaises(TypeError): - ra.get_content_type() - with self.assertRaises(TypeError): - ra.get_str('foo') - with self.assertRaises(TypeError): - ra.get_bytes('foo') - with self.assertRaises(TypeError): - ra.get_content_type('foo') - - def test_requestarguments_index(self): - """ - Make sure indexing a RequestArguments instance creates a new entry on the fly. - """ - ra = RequestArguments() - self.assertEqual(0, len(ra)) - self.assertFalse('foo' in ra) - # Create new entry - _ = ra['foo'] - self.assertEqual(1, len(ra)) - self.assertTrue('foo' in ra) - # Already exists, no new entry created - _ = ra['foo'] - self.assertEqual(1, len(ra)) - # Entry must be empty and mutable, and have the correct name - self.assertFalse(ra['foo'].is_view) - self.assertEqual(0, len(ra['foo'])) - self.assertEqual('foo', ra['foo'].name) - # Key must be a string - with self.assertRaises(TypeError): - # noinspection PyTypeChecker - _ = ra[42] - - def test_requestarguments_attr(self): - """ - Test attribute access syntactic sugar. - """ - ra = RequestArguments() - # Attribute should not exist yet - with self.assertRaises(KeyError): - _ = ra.foo - # Create entry - _ = ra['foo'] - # Creating entry should have created the attribute - self.assertEqual('foo', ra.foo.name) - # Attribute access should yield an immutable view - self.assertTrue(ra.foo.is_view) - - def test_requestarguments_iterate(self): - """ - Test iterating a RequestArguments instance. - """ - # Create an instance with some values - ra = RequestArguments() - ra['foo'].append('a', 'b') - ra['bar'].append('c', 'd') - ra['foo'].append('e', 'f') - # Container for test values (name, value) - items: Set[Tuple[str, str]] = set() - # Iterate RequestArguments instance, adding the name and value of each to the set - for a in ra: - items.add((a.name, str(a))) - # Compare result with expected value - self.assertEqual(2, len(items)) - self.assertIn(('foo', 'b'), items) - self.assertIn(('bar', 'd'), items) - - def test_requestarguments_full_use_case(self): - """ - Simulate a minimal RequestArguments use case. - """ - # Create empty RequestArguments instance - ra = RequestArguments() - # Parse GET request - getargs: Dict[str, List[str]] = urllib.parse.parse_qs('foo=42&bar=1337&foo=43&baz=Hello,%20World!') - # Insert GET arguments into RequestArguments - for k, vs in getargs.items(): - for v in vs: - ra[k].append('text/plain', v) - # Parse POST request - postargs: Dict[str, List[str]] = urllib.parse.parse_qs('foo=postfoo&postbar=42&foo=postfoo') - # Insert POST arguments into RequestArguments - for k, vs in postargs.items(): - # In this implementation, POST args replace GET args - ra[k].clear() - for v in vs: - ra[k].append('text/plain', v) - - # Someplace else: Use the RequestArguments instance. - self.assertEqual('1337', ra.bar.get_str()) - self.assertEqual('Hello, World!', ra.baz.get_str()) - self.assertEqual('42', ra.postbar.get_str()) - for a in ra.foo: - self.assertEqual('postfoo', a.get_str()) diff --git a/matemat/webserver/test/test_serve.py b/matemat/webserver/test/test_serve.py deleted file mode 100644 index e0caaa3..0000000 --- a/matemat/webserver/test/test_serve.py +++ /dev/null @@ -1,332 +0,0 @@ - -from typing import Any, Dict, Union - -import os -import os.path -from datetime import datetime, timedelta - -from matemat.exceptions import HttpException -from matemat.webserver import HttpHandler, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse -from matemat.webserver.test.abstract_httpd_test import AbstractHttpdTest, test_pagelet - - -@test_pagelet('/just/testing/serve_pagelet_str') -def serve_test_pagelet_str(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]: - return 'serve test pagelet str' - - -@test_pagelet('/just/testing/serve_pagelet_bytes') -def serve_test_pagelet_bytes(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]: - headers['Content-Type'] = 'application/x-foo-bar' - return b'serve\x80test\xffpagelet\xfebytes' - - -@test_pagelet('/just/testing/serve_pagelet_redirect') -def serve_test_pagelet_redirect(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]: - return RedirectResponse('/foo/bar') - - -@test_pagelet('/just/testing/serve_pagelet_template') -def serve_test_pagelet_template(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]: - return TemplateResponse('test.txt', what='World') - - -# noinspection PyTypeChecker -@test_pagelet('/just/testing/serve_pagelet_fail') -def serve_test_pagelet_fail(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]: - session_vars['test'] = 'hello, world!' - raise HttpException(599, 'Error expected during unit testing') - - -# noinspection PyTypeChecker -@test_pagelet('/just/testing/serve_pagelet_empty') -def serve_test_pagelet_empty(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]: - return PageletResponse() - - -# noinspection PyTypeChecker -@test_pagelet('/just/testing/serve_pagelet_type_error') -def serve_test_pagelet_fail(method: str, - path: str, - args: RequestArguments, - session_vars: Dict[str, Any], - headers: Dict[str, str], - pagelet_variables: Dict[str, str]) -> Union[bytes, str, PageletResponse]: - return 42 - - -class TestServe(AbstractHttpdTest): - """ - Test cases for the content serving of the web server. - """ - - def setUp(self): - super().setUp() - # Create a static resource in the temp dir - with open(os.path.join(self.tempdir.name, 'static_resource.txt'), 'w') as f: - f.write('static resource test') - # Create a second static resource chmodded to 0000, to test 403 Forbidden error - forbidden: str = os.path.join(self.tempdir.name, 'forbidden_static_resource.txt') - with open(forbidden, 'w') as f: - f.write('This should not be readable') - os.chmod(forbidden, 0) - # Create a CSS resource whose MIME type should be detected by file extension - with open(os.path.join(self.tempdir.name, 'teststyle.css'), 'w') as f: - f.write('.ninja { display: none; }\n') - # Create a file without extension (containing UTF-16 text with BOM); libmagic should take over - with open(os.path.join(self.tempdir.name, 'testdata'), 'wb') as f: - f.write(b'\xFE\xFFH\x00e\x00l\x00l\x00o\x00,\x00 \x00w\x00o\x00r\x00l\x00d\x00!\x00\n\x00') - # Create a file that will yield "text/plain; charset=binary" - with open(os.path.join(self.tempdir.name, 'testbin.txt'), 'wb') as f: - f.write(b'\x00\x00\x00\x00\x00\x00\x00\x00') - - def test_serve_pagelet_str(self): - # Call the test pagelet that produces a 200 OK result - self.client_sock.set_request(b'GET /just/testing/serve_pagelet_str HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure the correct pagelet was called - self.assertEqual('serve_test_pagelet_str', packet.pagelet) - # Make sure the expected content is served - self.assertEqual(200, packet.statuscode) - self.assertEqual(b'serve test pagelet str', packet.body) - - def test_serve_pagelet_bytes(self): - # Call the test pagelet that produces a 200 OK result - self.client_sock.set_request(b'GET /just/testing/serve_pagelet_bytes HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure the correct pagelet was called - self.assertEqual('serve_test_pagelet_bytes', packet.pagelet) - # Make sure the expected content is served - self.assertEqual(200, packet.statuscode) - self.assertEqual(b'serve\x80test\xffpagelet\xfebytes', packet.body) - - def test_serve_pagelet_fail(self): - # Call the test pagelet that produces a 500 Internal Server Error result - self.client_sock.set_request(b'GET /just/testing/serve_pagelet_fail HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure an error is raised - self.assertEqual(599, packet.statuscode) - - def test_serve_pagelet_redirect(self): - # Call the test pagelet that redirects to another path - self.client_sock.set_request(b'GET /just/testing/serve_pagelet_redirect HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure the correct pagelet was called - self.assertEqual('serve_test_pagelet_redirect', packet.pagelet) - # Make sure the correct redirect is issued - self.assertEqual(301, packet.statuscode) - self.assertEqual('/foo/bar', packet.headers['Location']) - # Make sure the response body is empty - self.assertEqual(0, len(packet.body)) - - def test_serve_pagelet_template(self): - # Call the test pagelet that redirects to another path - self.client_sock.set_request(b'GET /just/testing/serve_pagelet_template HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure the correct pagelet was called - self.assertEqual('serve_test_pagelet_template', packet.pagelet) - self.assertEqual(200, packet.statuscode) - # Make sure the response body was rendered correctly by the templating engine - self.assertEqual(b'Hello, World!', packet.body) - - def test_serve_pagelet_empty(self): - # Call the test pagelet that redirects to another path - self.client_sock.set_request(b'GET /just/testing/serve_pagelet_empty HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure the correct pagelet was called - self.assertEqual('serve_test_pagelet_empty', packet.pagelet) - self.assertEqual(200, packet.statuscode) - # Make sure the response body was rendered correctly by the templating engine - self.assertEqual(b'', packet.body) - - def test_serve_static_ok(self): - # Request a static resource - self.client_sock.set_request(b'GET /static_resource.txt HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure that no pagelet was called - self.assertIsNone(packet.pagelet) - # Make sure the expected content is served - self.assertEqual(200, packet.statuscode) - self.assertEqual(b'static resource test', packet.body) - - def test_serve_static_forbidden(self): - # Request a static resource with lacking permissions - self.client_sock.set_request(b'GET /forbidden_static_resource.txt HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure that no pagelet was called - self.assertIsNone(packet.pagelet) - # Make sure a 403 header is served - self.assertEqual(403, packet.statuscode) - self.assertNotEqual(b'This should not be readable', packet.body) - - def test_serve_static_cache(self): - # Request a static resource - timeout: datetime = datetime.utcnow() + timedelta(hours=1) - timeoutstr = timeout.strftime('%a, %d %b %Y %H:%M:%S GMT') - self.client_sock.set_request( - f'GET /static_resource.txt HTTP/1.1\r\nIf-Modified-Since: {timeoutstr}\r\n\r\n'.encode('utf-8')) - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure that no pagelet was called - self.assertIsNone(packet.pagelet) - # Make sure a 304 Not Modified is sent and the body is empty - self.assertEqual(304, packet.statuscode) - self.assertEqual(0, len(packet.body)) - - def test_serve_static_cache_renew(self): - # Request a static resource - self.client_sock.set_request( - b'GET /static_resource.txt HTTP/1.1\r\nIf-Modified-Since: Mon, 01 Jan 2018 13:37:42 GMT\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure that no pagelet was called - self.assertIsNone(packet.pagelet) - # Make sure the expected content is served - self.assertEqual(200, packet.statuscode) - self.assertEqual(b'static resource test', packet.body) - - def test_serve_not_found(self): - # Request a nonexistent resource - self.client_sock.set_request(b'GET /nonexistent HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure that no pagelet was called - self.assertIsNone(packet.pagelet) - # Make sure a 404 header is served - self.assertEqual(404, packet.statuscode) - - def test_serve_directory_traversal(self): - # Request a resource outside the webroot - self.client_sock.set_request(b'GET /../../../../../etc/passwd HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure that no pagelet was called - self.assertIsNone(packet.pagelet) - # Make sure a 404 header is served - self.assertEqual(404, packet.statuscode) - - def test_static_post_not_allowed(self): - # Request a resource outside the webroot - self.client_sock.set_request(b'POST /iwanttopostthis HTTP/1.1\r\n' - b'Content-Type: application/x-www-form-urlencoded\r\n' - b'Content-length: 37\r\n\r\n' - b'q=this%20should%20not%20be%20uploaded') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - - # Make sure that no pagelet was called - self.assertIsNone(packet.pagelet) - # Make sure a 405 Method Not Allowed header is served - self.assertEqual(405, packet.statuscode) - - def test_serve_pagelet_libmagic(self): - # The correct Content-Type header must be guessed, if a pagelet does not provide one - self.client_sock.set_request(b'GET /just/testing/serve_pagelet_str HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - self.assertEqual('text/plain; charset=us-ascii', packet.headers['Content-Type']) - - def test_serve_pagelet_libmagic_skipped(self): - # The Content-Type set by a pagelet should not be overwritten - self.client_sock.set_request(b'GET /just/testing/serve_pagelet_bytes HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - self.assertEqual('application/x-foo-bar', packet.headers['Content-Type']) - - def test_serve_pagelet_type_error(self): - # A 500 error should be returned if a pagelet returns an invalid type - self.client_sock.set_request(b'GET /just/testing/serve_pagelet_type_error HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - # Make sure a 500 header is served - self.assertEqual(500, packet.statuscode) - - def test_serve_static_mime_extension(self): - # The correct Content-Type should be guessed by file extension primarily - self.client_sock.set_request(b'GET /teststyle.css HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - # libmagic would say text/plain instead - self.assertEqual('text/css; charset=us-ascii', packet.headers['Content-Type']) - - def test_serve_static_mime_magic(self): - # The correct Content-Type should be guessed by file extension primarily - self.client_sock.set_request(b'GET /testdata HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - # Extension-based would fail, as there is no extension - self.assertEqual('text/plain; charset=utf-16be', packet.headers['Content-Type']) - - def test_serve_static_mime_magic_binary(self): - # The correct Content-Type should be guessed by file extension primarily - self.client_sock.set_request(b'GET /testbin.txt HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - # No charset should be in the header. Yes, this is a stupid example - self.assertEqual('text/plain', packet.headers['Content-Type']) - - def test_serve_static_pagelet_header(self): - # Send a request - self.client_sock.set_request(b'GET /just/testing/serve_pagelet_str HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - # The statically set header must be present - self.assertEqual('helloworld!', packet.headers['X-Static-Testing-Header']) - - def test_serve_static_static_header(self): - # Send a request - self.client_sock.set_request(b'GET /testbin.txt HTTP/1.1\r\n\r\n') - HttpHandler(self.client_sock, ('::1', 45678), self.server) - packet = self.client_sock.get_response() - # The statically set header must be present - self.assertEqual('helloworld!', packet.headers['X-Static-Testing-Header']) diff --git a/matemat/webserver/util.py b/matemat/webserver/util.py deleted file mode 100644 index c95d303..0000000 --- a/matemat/webserver/util.py +++ /dev/null @@ -1,138 +0,0 @@ - -from typing import Dict, List, Tuple, Optional - -import urllib.parse - -from matemat.webserver import RequestArguments, RequestArgument - - -def _parse_multipart(body: bytes, boundary: str) -> List[RequestArgument]: - """ - Given a HTTP body with form-data in multipart form, and the multipart-boundary, parse the multipart items and - return them as a dictionary. - - :param body: The HTTP multipart/form-data body. - :param boundary: The multipart boundary. - :return: A dictionary of field names as key, and content types and field values as value. - """ - # Prepend a CRLF for the first boundary to match - body = b'\r\n' + body - # Generate item header boundary and terminating boundary from general boundary string - _boundary = f'\r\n--{boundary}\r\n'.encode('utf-8') - _end_boundary = f'\r\n--{boundary}--\r\n'.encode('utf-8') - # Split at the end boundary and make sure there comes nothing after it - allparts = body.split(_end_boundary, 1) - if len(allparts) != 2 or allparts[1] != b'': - raise ValueError('Last boundary missing or corrupted') - # Split remaining body into its parts, and verify at least 1 part is there - parts: List[bytes] = (allparts[0]).split(_boundary) - if len(parts) < 1 or parts[0] != b'': - raise ValueError('First boundary missing or corrupted') - # Remove the first, empty part - parts = parts[1:] - - # Results go into this dict - args: Dict[str, RequestArgument] = dict() - - # Parse each multipart part - for part in parts: - # Parse multipart headers - hdr: Dict[str, str] = dict() - while True: - head, part = part.split(b'\r\n', 1) - # Break on header/body delimiter - if head == b'': - break - # Add header to hdr dict - hk, hv = head.decode('utf-8').split(':') - hdr[hk.strip()] = hv.strip() - # No content type set - set broadest possible type - if 'Content-Type' not in hdr: - hdr['Content-Type'] = 'application/octet-stream' - # At least Content-Disposition must be present - if 'Content-Disposition' not in hdr: - raise ValueError('Missing Content-Type or Content-Disposition header') - # Extract Content-Disposition header value and its arguments - cd, *cdargs = hdr['Content-Disposition'].split(';') - # Content-Disposition MUST be form-data; everything else is rejected - if cd.strip() != 'form-data': - raise ValueError(f'Unknown Content-Disposition: {cd}') - # Extract the "name" header argument - has_name = False - for cdarg in cdargs: - k, v = cdarg.split('=', 1) - if k.strip() == 'name': - has_name = True - name: str = v.strip() - # Remove quotation marks around the name value - if name.startswith('"') and name.endswith('"'): - name = v[1:-1] - # Add the Content-Type and the content to the header, with the provided name - if name not in args: - args[name] = RequestArgument(name) - args[name].append(hdr['Content-Type'].strip(), part) - if not has_name: - # Content-Disposition header without name attribute - raise ValueError('mutlipart/form-data part without name attribute') - - return list(args.values()) - - -def parse_args(request: str, postbody: Optional[bytes] = None, enctype: str = 'text/plain') \ - -> Tuple[str, RequestArguments]: - """ - Given a HTTP request path, and optionally a HTTP POST body in application/x-www-form-urlencoded or - multipart/form-data form, parse the arguments and return them as a dictionary. - - If a key is used both in GET and in POST, the POST value takes precedence, and the GET value is discarded. - - :param request: The request string to parse. - :param postbody: The POST body to parse, defaults to None. - :param enctype: Encoding of the POST body; supported values are application/x-www-form-urlencoded and - multipart/form-data. - :return: A tuple consisting of the base path and a dictionary with the parsed key/value pairs, and the value's - content type. - """ - # Parse the request "URL" (i.e. only the path) - tokens = urllib.parse.urlparse(request) - # Parse the GET arguments - if len(tokens.query) == 0: - getargs: Dict[str, List[str]] = dict() - else: - getargs = urllib.parse.parse_qs(tokens.query, strict_parsing=True, keep_blank_values=True, errors='strict') - - args = RequestArguments() - for k, vs in getargs.items(): - args[k].clear() - for v in vs: - args[k].append('text/plain', v) - - if postbody is not None: - if enctype == 'application/x-www-form-urlencoded': - # Parse the POST body - pb: str = postbody.decode('utf-8') - if len(pb) == 0: - postargs: Dict[str, List[str]] = dict() - else: - postargs = urllib.parse.parse_qs(pb, strict_parsing=True, keep_blank_values=True, errors='strict') - # Write all POST values into the dict, overriding potential duplicates from GET - for k, vs in postargs.items(): - args[k].clear() - for v in vs: - args[k].append('text/plain', v) - elif enctype.startswith('multipart/form-data'): - # Parse the multipart boundary from the Content-Type header - try: - boundary: str = enctype.split('boundary=')[1].strip() - except IndexError: - raise ValueError('Multipart boundary in header not set or corrupted') - # Parse the multipart body - mpargs = _parse_multipart(postbody, boundary) - for ra in mpargs: - args[ra.name].clear() - for a in ra: - args[ra.name].append(a.get_content_type(), bytes(a)) - else: - raise ValueError(f'Unsupported Content-Type: {enctype}') - # Return the path and the parsed arguments - return tokens.path, args diff --git a/setup.py b/setup.py index 6c2d005..c135a73 100644 --- a/setup.py +++ b/setup.py @@ -19,6 +19,7 @@ soda machine's touch screen). packages=find_packages(exclude=['*.test']), python_requires='>=3.6', install_requires=[ + 'bottle', 'file-magic', 'jinja2', 'Pillow' diff --git a/templates/admin_all.html b/templates/admin_all.html index d1aa1f2..55a2cc6 100644 --- a/templates/admin_all.html +++ b/templates/admin_all.html @@ -70,7 +70,7 @@ - + diff --git a/templates/base.html b/templates/base.html index a33f411..414f711 100644 --- a/templates/base.html +++ b/templates/base.html @@ -5,7 +5,7 @@ {% block head %} {# Show the setup name, as set in the config file, as tab title. Don't escape HTML entities. #} {{ setupname|safe }} - + {% endblock %} diff --git a/templates/touchkey.html b/templates/touchkey.html index 9d14244..0d4934f 100644 --- a/templates/touchkey.html +++ b/templates/touchkey.html @@ -26,7 +26,7 @@ Cancel - +