1
0
Fork 0
forked from s3lph/matemat

Migrate from own web framework to bottle.py

This commit is contained in:
s3lph 2019-11-22 03:58:54 +01:00
parent 9b1d8eb61c
commit 8e8f159150
36 changed files with 580 additions and 3067 deletions

View file

@ -1,27 +1,87 @@
from typing import Any, Dict, Iterable, Union from typing import Any, Dict, Iterable, Union
from bottle import route, static_file, run
import sys import sys
import os.path
from matemat.webserver import MatematWebserver from matemat.db import MatematDatabase
from matemat.webserver import parse_config_file 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. # Those imports are actually needed, as they implicitly register pagelets.
# noinspection PyUnresolvedReferences # noinspection PyUnresolvedReferences
from matemat.webserver.pagelets import * 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/<filename:path>')
def serve_static_files(filename: str):
config = get_config()
staticroot = os.path.abspath(config['staticroot'])
return static_file(filename, root=staticroot)
def main(): 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' configfile: Union[str, Iterable[str]] = '/etc/matemat.conf'
if len(sys.argv) > 1: if len(sys.argv) > 1:
configfile = sys.argv[1:] configfile = sys.argv[1:]
parse_config_file(configfile)
# Parse the config file config = get_config()
config: Dict[str, Any] = parse_config_file(configfile)
_init(config)
host: str = config['listen']
port: int = int(str(config['port']))
# Start the web server # Start the web server
MatematWebserver(**config).start() run(host=host, port=port, debug=True)
if __name__ == '__main__': if __name__ == '__main__':

View file

@ -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. server will attempt to serve the request with a static resource in a previously configured webroot directory.
""" """
from .requestargs import RequestArgument, RequestArguments from .logger import Logger
from .responses import PageletResponse, RedirectResponse, TemplateResponse
from .httpd import MatematWebserver, HttpHandler, pagelet, pagelet_init, pagelet_cron
from .config import parse_config_file

View file

@ -7,6 +7,9 @@ import logging
from configparser import ConfigParser from configparser import ConfigParser
config: Dict[str, Any] = dict()
def parse_logging(symbolic_level: str, symbolic_target: str) -> Tuple[int, logging.Handler]: def parse_logging(symbolic_level: str, symbolic_target: str) -> Tuple[int, logging.Handler]:
""" """
Parse the LogLevel and LogTarget from configuration. 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 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. Parse the configuration file at the given path.
:param paths: The config file(s) to parse. :param paths: The config file(s) to parse.
:return: A dictionary containing the parsed configuration. :return: A dictionary containing the parsed configuration.
""" """
global config
# Set up default values # Set up default values
config: Dict[str, Any] = {
# Address to listen on # Address to listen on
'listen': '::', config['listen'] = '::'
# TCP port to listen on # TCP port to listen on
'port': 80, config['port'] = 80
# Root directory of statically served content # Root directory of statically served content
'staticroot': '/var/matemat/static', config['staticroot'] = '/var/matemat/static'
# Root directory of Jinja2 templates # Root directory of Jinja2 templates
'templateroot': '/var/matemat/templates', config['templateroot'] = '/var/matemat/templates'
# Log level # Log level
'log_level': logging.INFO, config['log_level'] = logging.INFO
# Log target: An IO stream (stderr, stdout, ...) or a filename # Log target: An IO stream (stderr, stdout, ...) or a filename
'log_handler': logging.StreamHandler(), config['log_handler'] = logging.StreamHandler()
# Variables passed to pagelets # Variables passed to pagelets
'pagelet_variables': dict(), config['pagelet_variables'] = dict()
# Statically configured headers # Statically configured headers
'headers': dict() config['headers'] = dict()
}
# Initialize the config parser # Initialize the config parser
parser: ConfigParser = ConfigParser() 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(): for k, v in parser['HttpHeaders'].items():
config['headers'][k] = v config['headers'][k] = v
def get_config() -> Dict[str, Any]:
global config
return config return config
def get_app_config() -> Dict[str, Any]:
global config
return config['pagelet_variables']

85
matemat/webserver/cron.py Normal file
View file

@ -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

View file

@ -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)

View file

@ -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

View file

@ -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. 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 .main import main_page
from .login import login_page from .login import login_page
from .logout import logout from .logout import logout

View file

@ -1,69 +1,67 @@
from typing import Any, Dict, Union
import os import os
from shutil import copyfile
import magic
from io import BytesIO 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 import MatematDatabase
from matemat.db.primitives import User, ReceiptPreference 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') @get('/admin')
def admin(method: str, @post('/admin')
path: str, def admin():
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
config: Dict[str, str]) \
-> Union[str, bytes, PageletResponse]:
""" """
The admin panel, shows a user's own settings. Additionally, for administrators, settings to modify other users and The admin panel, shows a user's own settings. Additionally, for administrators, settings to modify other users and
products are shown. products are shown.
""" """
config = get_app_config()
session_id: str = session.start()
# If no user is logged in, redirect to the login page # 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: if not session.has(session_id, 'authentication_level') or not session.has(session_id, 'authenticated_user'):
return RedirectResponse('/login') redirect('/login')
authlevel: int = session_vars['authentication_level'] authlevel: int = session.get(session_id, 'authentication_level')
uid: int = session_vars['authenticated_user'] 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) # Show a 403 Forbidden error page if no user is logged in (0) or a user logged in via touchkey (1)
if authlevel < 2: if authlevel < 2:
raise HttpException(403) abort(403)
# Connect to the database # Connect to the database
with MatematDatabase(config['DatabaseFile']) as db: with MatematDatabase(config['DatabaseFile']) as db:
# Fetch the authenticated user # Fetch the authenticated user
user = db.get_user(uid) user = db.get_user(uid)
# If the POST request contains a "change" parameter, delegate the change handling to the function below # If the POST request contains a "change" parameter, delegate the change handling to the function below
if method == 'POST' and 'change' in args: if request.method == 'POST' and 'change' in request.params:
handle_change(args, user, db, config) handle_change(request.params, request.files, user, db)
# If the POST request contains an "adminchange" parameter, delegate the change handling to the function below # 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: elif request.method == 'POST' and 'adminchange' in request.params and user.is_admin:
handle_admin_change(args, db, config) handle_admin_change(request.params, request.files, db)
# Fetch all existing users and products from the database # Fetch all existing users and products from the database
users = db.list_users() users = db.list_users()
products = db.list_products() products = db.list_products()
# Render the "Admin/Settings" page # Render the "Admin/Settings" page
return TemplateResponse('admin.html', return template.render('admin.html',
authuser=user, authlevel=authlevel, users=users, products=products, authuser=user, authlevel=authlevel, users=users, products=products,
receipt_preference_class=ReceiptPreference, receipt_preference_class=ReceiptPreference,
setupname=config['InstanceName'], config_smtp_enabled=config['SmtpSendReceipts']) 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. 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 user: The user to edit.
:param db: The database facade where changes are written to. :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: try:
# Read the type of change requested by the user, then switch over it # Read the type of change requested by the user, then switch over it
change = str(args.change) change = str(args.change)
@ -122,10 +120,10 @@ def handle_change(args: RequestArguments, user: User, db: MatematDatabase, confi
# The user requested an avatar change # The user requested an avatar change
elif change == 'avatar': elif change == 'avatar':
# The new avatar field must be present # The new avatar field must be present
if 'avatar' not in args: if 'avatar' not in files:
return return
# Read the raw image data from the request # 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 # Only process the image, if its size is more than zero. Zero size means no new image was uploaded
if len(avatar) == 0: if len(avatar) == 0:
return return
@ -150,14 +148,14 @@ def handle_change(args: RequestArguments, user: User, db: MatematDatabase, confi
raise ValueError('an argument not a string') 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. Write the changes requested by an admin for users of products.
:param args: The RequestArguments object passed to the pagelet. :param args: The RequestArguments object passed to the pagelet.
:param db: The database facade where changes are written to. :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: try:
# Read the type of change requested by the admin, then switch over it # Read the type of change requested by the admin, then switch over it
change = str(args.adminchange) change = str(args.adminchange)
@ -202,11 +200,10 @@ def handle_admin_change(args: RequestArguments, db: MatematDatabase, config: Dic
# Create the user in the database # Create the user in the database
newproduct = db.create_product(name, price_member, price_non_member) newproduct = db.create_product(name, price_member, price_non_member)
# If a new product image was uploaded, process it # If a new product image was uploaded, process it
if 'image' in args and len(bytes(args.image)) > 0: image = files.image.file.read() if 'image' in files else None
# Read the raw image data from the request if image is None or len(image) == 0:
avatar = bytes(args.image)
# Detect the MIME type # 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/'): if not filemagic.mime_type.startswith('image/'):
return return
# Create the absolute path of the upload directory # 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) os.makedirs(abspath, exist_ok=True)
try: try:
# Parse the image data # Parse the image data
image: Image = Image.open(BytesIO(avatar)) image: Image = Image.open(BytesIO(image))
# Resize the image to 150x150 # Resize the image to 150x150
image.thumbnail((150, 150), Image.LANCZOS) image.thumbnail((150, 150), Image.LANCZOS)
# Write the image to the file # 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: if category not in args:
continue continue
# Read the raw image data from the request # 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 # Only process the image, if its size is more than zero. Zero size means no new image was uploaded
if len(default) == 0: if len(default) == 0:
continue continue

View file

@ -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.db import MatematDatabase
from matemat.webserver import session
from matemat.webserver.config import get_app_config
@pagelet('/buy') @get('/buy')
def buy(method: str, @post('/buy')
path: str, def buy():
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
config: Dict[str, str]) \
-> Union[str, bytes, PageletResponse]:
""" """
The purchasing mechanism. Called by the user clicking an item on the product list. 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 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: if not session.has(session_id, 'authenticated_user'):
return RedirectResponse('/') redirect('/')
# Connect to the database # Connect to the database
with MatematDatabase(config['DatabaseFile']) as db: with MatematDatabase(config['DatabaseFile']) as db:
# Fetch the authenticated user from the database # 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) user = db.get_user(uid)
# Read the product from the database, identified by the product ID passed as request argument # Read the product from the database, identified by the product ID passed as request argument
if 'pid' in args: if 'pid' in request.params:
pid = int(str(args.pid)) pid = int(str(request.params.pid))
product = db.get_product(pid) product = db.get_product(pid)
# Create a consumption entry for the (user, product) combination # Create a consumption entry for the (user, product) combination
db.increment_consumption(user, product) db.increment_consumption(user, product)
# Redirect to the main page (where this request should have come from) # Redirect to the main page (where this request should have come from)
return RedirectResponse('/') redirect('/')

View file

@ -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.db import MatematDatabase
from matemat.webserver import session
from matemat.webserver.config import get_app_config
@pagelet('/deposit') @get('/deposit')
def deposit(method: str, @post('/deposit')
path: str, def deposit():
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
config: Dict[str, str]) \
-> Union[str, bytes, PageletResponse]:
""" """
The cash depositing mechanism. Called by the user submitting a deposit from the product list. 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 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: if not session.has(session_id, 'authenticated_user'):
return RedirectResponse('/') redirect('/')
# Connect to the database # Connect to the database
with MatematDatabase(config['DatabaseFile']) as db: with MatematDatabase(config['DatabaseFile']) as db:
# Fetch the authenticated user from the database # 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) 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 # 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 # Write the deposit to the database
db.deposit(user, n) db.deposit(user, n)
# Redirect to the main page (where this request should have come from) # Redirect to the main page (where this request should have come from)
return RedirectResponse('/') redirect('/')

View file

@ -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

View file

@ -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 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') @get('/login')
def login_page(method: str, @post('/login')
path: str, def login_page():
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
config: Dict[str, str]) \
-> Union[bytes, str, PageletResponse]:
""" """
The password login mechanism. If called via GET, render the UI template; if called via POST, attempt to log in with 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). 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 a user is already logged in, simply redirect to the main page, showing the product list
if 'authenticated_user' in session_vars: if session.has(session_id, 'authenticated_user'):
return RedirectResponse('/') redirect('/')
# If requested via HTTP GET, render the login page showing the login UI # If requested via HTTP GET, render the login page showing the login UI
if method == 'GET': if request.method == 'GET':
return TemplateResponse('login.html', return template.render('login.html',
setupname=config['InstanceName']) setupname=config['InstanceName'])
# If requested via HTTP POST, read the request arguments and attempt to log in with the provided credentials # 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 # Connect to the database
with MatematDatabase(config['DatabaseFile']) as db: with MatematDatabase(config['DatabaseFile']) as db:
try: try:
# Read the request arguments and attempt to log in with them # 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: except AuthenticationError:
# Reload the touchkey login page on failure # Reload the touchkey login page on failure
return RedirectResponse('/login') redirect('/login')
# Set the user ID session variable # 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) # 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 # 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 # If neither GET nor POST was used, show a 405 Method Not Allowed error page
raise HttpException(405) abort(405)

View file

@ -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') @get('/logout')
def logout(method: str, @post('/logout')
path: str, def logout():
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
config: Dict[str, str]) \
-> Union[bytes, str, PageletResponse]:
""" """
The logout mechanism, clearing the authentication values in the session storage. 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 # Remove the authenticated user ID from the session storage, if any
if 'authenticated_user' in session_vars: if session.has(session_id, 'authenticated_user'):
del session_vars['authenticated_user'] session.delete(session_id, 'authenticated_user')
# Reset the authlevel session variable (0 = none, 1 = touchkey, 2 = password login) # 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 # Redirect to the main page, showing the user list
return RedirectResponse('/') redirect('/')

View file

@ -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.db import MatematDatabase
from matemat.webserver import template, session
from matemat.webserver.config import get_app_config
@pagelet('/') @route('/')
def main_page(method: str, def main_page():
path: str,
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
config: Dict[str, str]) \
-> Union[bytes, str, PageletResponse]:
""" """
The main page, showing either the user list (if no user is logged in) or the product list (if a user is logged in). 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: with MatematDatabase(config['DatabaseFile']) as db:
# Check whether a user is logged in # 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 # Fetch the user id and authentication level (touchkey vs password) from the session storage
uid: int = session_vars['authenticated_user'] uid: int = session.get(session_id, 'authenticated_user')
authlevel: int = session_vars['authentication_level'] authlevel: int = session.get(session_id, 'authentication_level')
# Fetch the user object from the database (for name display, price calculation and admin check) # Fetch the user object from the database (for name display, price calculation and admin check)
user = db.get_user(uid) user = db.get_user(uid)
# Fetch the list of products to display # Fetch the list of products to display
products = db.list_products() products = db.list_products()
# Prepare a response with a jinja2 template # Prepare a response with a jinja2 template
return TemplateResponse('productlist.html', return template.render('productlist.html',
authuser=user, products=products, authlevel=authlevel, authuser=user, products=products, authlevel=authlevel,
setupname=config['InstanceName']) setupname=config['InstanceName'])
else: else:
# If there are no admin users registered, jump to the admin creation procedure # If there are no admin users registered, jump to the admin creation procedure
if not db.has_admin_users(): 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 # If no user is logged in, fetch the list of users and render the userlist template
users = db.list_users(with_touchkey=True) users = db.list_users(with_touchkey=True)
return TemplateResponse('userlist.html', return template.render('userlist.html',
users=users, setupname=config['InstanceName']) users=users, setupname=config['InstanceName'])

View file

@ -1,36 +1,35 @@
from typing import Any, Dict, Union
import os import os
from io import BytesIO
from typing import Dict
import magic import magic
from PIL import Image 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 import MatematDatabase
from matemat.db.primitives import Product 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.util.currency_format import parse_chf
from matemat.webserver import template, session
from matemat.webserver.config import get_app_config
@pagelet('/modproduct') @get('/modproduct')
def modproduct(method: str, @post('/modproduct')
path: str, def modproduct():
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
config: Dict[str, str]) \
-> Union[str, bytes, PageletResponse]:
""" """
The product modification page available from the admin panel. 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 no user is logged in, redirect to the login page
if 'authentication_level' not in session_vars or 'authenticated_user' not in session_vars: if not session.has(session_id, 'authentication_level') or not session.has(session_id, 'authenticated_user'):
return RedirectResponse('/login') redirect('/login')
authlevel: int = session_vars['authentication_level'] authlevel: int = session.get(session_id, 'authentication_level')
auth_uid: int = session_vars['authenticated_user'] 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) # Show a 403 Forbidden error page if no user is logged in (0) or a user logged in via touchkey (1)
if authlevel < 2: if authlevel < 2:
raise HttpException(403) abort(403)
# Connect to the database # Connect to the database
with MatematDatabase(config['DatabaseFile']) as db: with MatematDatabase(config['DatabaseFile']) as db:
@ -38,38 +37,38 @@ def modproduct(method: str,
authuser = db.get_user(auth_uid) authuser = db.get_user(auth_uid)
if not authuser.is_admin: if not authuser.is_admin:
# Show a 403 Forbidden error page if the user is not an admin # Show a 403 Forbidden error page if the user is not an admin
raise HttpException(403) abort(403)
if 'productid' not in args: if 'productid' not in request.params:
# Show a 400 Bad Request error page if no product to edit was specified # Show a 400 Bad Request error page if no product to edit was specified
# (should never happen during normal operation) # (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 # 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) product = db.get_product(modproduct_id)
# If the request contains a "change" parameter, delegate the change handling to the function below # If the request contains a "change" parameter, delegate the change handling to the function below
if 'change' in args: if 'change' in request.params:
handle_change(args, product, db, config) 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 the product was deleted, redirect back to the admin page, as there is nothing to edit any more
if str(args.change) == 'del': if str(request.params.change) == 'del':
return RedirectResponse('/admin') redirect('/admin')
# Render the "Modify Product" page # Render the "Modify Product" page
return TemplateResponse('modproduct.html', return template.render('modproduct.html',
authuser=authuser, product=product, authlevel=authlevel, authuser=authuser, product=product, authlevel=authlevel,
setupname=config['InstanceName']) 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. 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 product: The product to edit.
:param db: The database facade where changes are written to. :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 # Read the type of change requested by the admin, then switch over it
change = str(args.change) change = str(args.change)
@ -101,9 +100,9 @@ def handle_change(args: RequestArguments, product: Product, db: MatematDatabase,
except DatabaseConsistencyError: except DatabaseConsistencyError:
return return
# If a new product image was uploaded, process it # 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 # 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 # Only process the image, if its size is more than zero. Zero size means no new image was uploaded
if len(avatar) == 0: if len(avatar) == 0:
return return

View file

@ -1,36 +1,35 @@
from typing import Any, Dict, Optional, Union
import os import os
from io import BytesIO
from typing import Dict, Optional
import magic import magic
from PIL import Image 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 import MatematDatabase
from matemat.db.primitives import User, ReceiptPreference 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.util.currency_format import parse_chf
from matemat.webserver import template, session
from matemat.webserver.config import get_app_config
@pagelet('/moduser') @get('/moduser')
def moduser(method: str, @post('/moduser')
path: str, def moduser():
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
config: Dict[str, str]) \
-> Union[str, bytes, PageletResponse]:
""" """
The user modification page available from the admin panel. 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 no user is logged in, redirect to the login page
if 'authentication_level' not in session_vars or 'authenticated_user' not in session_vars: if not session.has(session_id, 'authentication_level') or not session.has(session_id, 'authenticated_user'):
return RedirectResponse('/login') redirect('/login')
authlevel: int = session_vars['authentication_level'] authlevel: int = session.get(session_id, 'authentication_level')
auth_uid: int = session_vars['authenticated_user'] 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) # Show a 403 Forbidden error page if no user is logged in (0) or a user logged in via touchkey (1)
if authlevel < 2: if authlevel < 2:
raise HttpException(403) abort(403)
# Connect to the database # Connect to the database
with MatematDatabase(config['DatabaseFile']) as db: with MatematDatabase(config['DatabaseFile']) as db:
@ -38,31 +37,31 @@ def moduser(method: str,
authuser = db.get_user(auth_uid) authuser = db.get_user(auth_uid)
if not authuser.is_admin: if not authuser.is_admin:
# Show a 403 Forbidden error page if the user is not an admin # Show a 403 Forbidden error page if the user is not an admin
raise HttpException(403) abort(403)
if 'userid' not in args: if 'userid' not in request.params:
# Show a 400 Bad Request error page if no users to edit was specified # Show a 400 Bad Request error page if no users to edit was specified
# (should never happen during normal operation) # (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 # 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) user = db.get_user(moduser_id)
# If the request contains a "change" parameter, delegate the change handling to the function below # If the request contains a "change" parameter, delegate the change handling to the function below
if 'change' in args: if 'change' in request.params:
handle_change(args, user, authuser, db, config) 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 the user was deleted, redirect back to the admin page, as there is nothing to edit any more
if str(args.change) == 'del': if str(request.params.change) == 'del':
return RedirectResponse('/admin') redirect('/admin')
# Render the "Modify User" page # Render the "Modify User" page
return TemplateResponse('moduser.html', return template.render('moduser.html',
authuser=authuser, user=user, authlevel=authlevel, authuser=authuser, user=user, authlevel=authlevel,
receipt_preference_class=ReceiptPreference, receipt_preference_class=ReceiptPreference,
setupname=config['InstanceName'], config_smtp_enabled=config['SmtpSendReceipts']) 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: -> None:
""" """
Write the changes requested by an admin to the database. 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 user: The user to edit.
:param authuser: The user performing the modification. :param authuser: The user performing the modification.
:param db: The database facade where changes are written to. :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 # Read the type of change requested by the admin, then switch over it
change = str(args.change) change = str(args.change)
@ -124,9 +123,9 @@ def handle_change(args: RequestArguments, user: User, authuser: User, db: Matema
except DatabaseConsistencyError: except DatabaseConsistencyError:
return return
# If a new avatar was uploaded, process it # If a new avatar was uploaded, process it
if 'avatar' in args: if 'avatar' in files:
# Read the raw image data from the request # 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 # Only process the image, if its size is more than zero. Zero size means no new image was uploaded
if len(avatar) == 0: if len(avatar) == 0:
return return

View file

@ -1,22 +1,25 @@
from typing import Dict, List, Tuple from typing import List, Tuple
import logging import logging
import smtplib as smtp import smtplib as smtp
from email.mime.multipart import MIMEMultipart from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText 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 import MatematDatabase
from matemat.db.primitives import User, Receipt from matemat.db.primitives import User, Receipt
from matemat.util.currency_format import format_chf from matemat.util.currency_format import format_chf
@pagelet_cron(hours=6) logger: logging.Logger = logging.Logger(__file__)
def receipt_smtp_cron(config: Dict[str, str],
jinja_env: Environment,
logger: logging.Logger) -> None: @cron(hours=6)
def receipt_smtp_cron() -> None:
config = get_app_config()
if config['SmtpSendReceipts'] != '1': if config['SmtpSendReceipts'] != '1':
# Sending receipts via mail is disabled # Sending receipts via mail is disabled
return return
@ -35,15 +38,13 @@ def receipt_smtp_cron(config: Dict[str, str],
receipts.append(receipt) receipts.append(receipt)
else: else:
logger.debug('No receipt due for user "%s".', user.name) 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: if len(receipts) > 0:
_send_receipt_mails(receipts, jinja_env, logger, config) _send_receipt_mails(receipts)
def _send_receipt_mails(receipts: List[Receipt], def _send_receipt_mails(receipts: List[Receipt]) -> None:
jinja_env: Environment, config = get_app_config()
logger: logging.Logger,
config: Dict[str, str]) -> None:
mails: List[Tuple[str, MIMEMultipart]] = [] mails: List[Tuple[str, MIMEMultipart]] = []
for receipt in receipts: for receipt in receipts:
if receipt.user.email is None: 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) fbal = format_chf(receipt.transactions[0].old_balance).rjust(12)
tbal: str = format_chf(receipt.user.balance).rjust(12) tbal: str = format_chf(receipt.user.balance).rjust(12)
# Render the receipt # Render the receipt
template: Template = jinja_env.get_template('receipt.txt') rendered: str = template.render('receipt.txt',
rendered: str = template.render(fdate=fdate, tdate=tdate, user=username, fbal=fbal, tbal=tbal, date=fdate, tdate=tdate, user=username, fbal=fbal, tbal=tbal,
receipt_id=receipt.id, transactions=receipt.transactions, receipt_id=receipt.id, transactions=receipt.transactions,
instance_name=config['InstanceName']) instance_name=config['InstanceName'])
# Put the rendered receipt in the message body # Put the rendered receipt in the message body

View file

@ -1,33 +1,30 @@
from typing import Any, Dict, List, Tuple, Union
from math import pi, sin, cos
from datetime import datetime, timedelta 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 import MatematDatabase
from matemat.db.primitives import User 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') @route('/statistics')
def statistics(method: str, def statistics():
path: str,
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
config: Dict[str, str]) \
-> Union[str, bytes, PageletResponse]:
""" """
The statistics page available from the admin panel. 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 no user is logged in, redirect to the login page
if 'authentication_level' not in session_vars or 'authenticated_user' not in session_vars: if not session.has(session_id, 'authentication_level') or not session.has(session_id, 'authenticated_user'):
return RedirectResponse('/login') redirect('/login')
authlevel: int = session_vars['authentication_level'] authlevel: int = session.get(session_id, 'authentication_level')
auth_uid: int = session_vars['authenticated_user'] 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) # Show a 403 Forbidden error page if no user is logged in (0) or a user logged in via touchkey (1)
if authlevel < 2: if authlevel < 2:
raise HttpException(403) abort(403)
# Connect to the database # Connect to the database
with MatematDatabase(config['DatabaseFile']) as db: with MatematDatabase(config['DatabaseFile']) as db:
@ -35,15 +32,15 @@ def statistics(method: str,
authuser: User = db.get_user(auth_uid) authuser: User = db.get_user(auth_uid)
if not authuser.is_admin: if not authuser.is_admin:
# Show a 403 Forbidden error page if the user is not an admin # Show a 403 Forbidden error page if the user is not an admin
raise HttpException(403) abort(403)
todate: datetime = datetime.utcnow() todate: datetime = datetime.utcnow()
fromdate: datetime = (todate - timedelta(days=365)).replace(hour=0, minute=0, second=0, microsecond=0) fromdate: datetime = (todate - timedelta(days=365)).replace(hour=0, minute=0, second=0, microsecond=0)
if 'fromdate' in args: if 'fromdate' in request.params:
fdarg: str = str(args.fromdate) fdarg: str = str(request.params.fromdate)
fromdate = datetime.strptime(fdarg, '%Y-%m-%d').replace(tzinfo=None) fromdate = datetime.strptime(fdarg, '%Y-%m-%d').replace(tzinfo=None)
if 'todate' in args: if 'todate' in request.params:
tdarg: str = str(args.todate) tdarg: str = str(request.params.todate)
todate = datetime.strptime(tdarg, '%Y-%m-%d').replace(tzinfo=None) todate = datetime.strptime(tdarg, '%Y-%m-%d').replace(tzinfo=None)
todate = (todate + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) todate = (todate + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
@ -70,7 +67,7 @@ def statistics(method: str,
previous_angle = angle previous_angle = angle
# Render the statistics page # Render the statistics page
return TemplateResponse('statistics.html', return template.render('statistics.html',
fromdate=fromdate.strftime('%Y-%m-%d'), fromdate=fromdate.strftime('%Y-%m-%d'),
todate=(todate - timedelta(days=1)).strftime('%Y-%m-%d'), todate=(todate - timedelta(days=1)).strftime('%Y-%m-%d'),
product_slices=slices, product_slices=slices,

View file

@ -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 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') @get('/touchkey')
def touchkey_page(method: str, @post('/touchkey')
path: str, def touchkey_page():
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
config: Dict[str, str]) \
-> Union[bytes, str, PageletResponse]:
""" """
The touchkey login mechanism. If called via GET, render the UI template; if called via POST, attempt to log in with 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). 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 a user is already logged in, simply redirect to the main page, showing the product list
if 'authenticated_user' in session_vars: if session.has(session_id, 'authenticated_user'):
return RedirectResponse('/') redirect('/')
# If requested via HTTP GET, render the login page showing the touchkey UI # If requested via HTTP GET, render the login page showing the touchkey UI
if method == 'GET': if request.method == 'GET':
return TemplateResponse('touchkey.html', return template.render('touchkey.html',
username=str(args.username), uid=int(str(args.uid)), username=str(request.params.username), uid=int(str(request.params.uid)),
setupname=config['InstanceName']) setupname=config['InstanceName'])
# If requested via HTTP POST, read the request arguments and attempt to log in with the provided credentials # 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 # Connect to the database
with MatematDatabase(config['DatabaseFile']) as db: with MatematDatabase(config['DatabaseFile']) as db:
try: try:
# Read the request arguments and attempt to log in with them # 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: except AuthenticationError:
# Reload the touchkey login page on failure # 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 # 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) # 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 # 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 # If neither GET nor POST was used, show a 405 Method Not Allowed error page
raise HttpException(405) abort(405)

View file

@ -1,43 +1,38 @@
from bottle import get, post, redirect, abort, request
from typing import Any, Dict, Union
from matemat.db import MatematDatabase from matemat.db import MatematDatabase
from matemat.webserver import pagelet, RequestArguments, PageletResponse, RedirectResponse, TemplateResponse from matemat.webserver import template
from matemat.exceptions import HttpException from matemat.webserver.config import get_app_config
@pagelet('/userbootstrap') @get('/userbootstrap')
def userbootstrap(method: str, @post('/userbootstrap')
path: str, def userbootstrap():
args: RequestArguments,
session_vars: Dict[str, Any],
headers: Dict[str, str],
config: Dict[str, str]) \
-> Union[bytes, str, PageletResponse]:
""" """
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 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. admin users left.
""" """
config = get_app_config()
with MatematDatabase(config['DatabaseFile']) as db: with MatematDatabase(config['DatabaseFile']) as db:
# Redirect to main if there are still administrators registered # Redirect to main if there are still administrators registered
if db.has_admin_users(): if db.has_admin_users():
return RedirectResponse('/') redirect('/')
# Process submission # Process submission
if method == 'POST': if request.method == 'POST':
# Make sure all required values are present # Make sure all required values are present
if 'username' not in args or 'password' not in args or 'password2' not in args: if 'username' not in request.params or 'password' not in request.params or 'password2' not in request.params:
raise HttpException(400, 'Some arguments are missing') abort(400, 'Some arguments are missing')
username: str = str(args.username) username: str = str(request.params.username)
password: str = str(args.password) password: str = str(request.params.password)
password2: str = str(args.password2) password2: str = str(request.params.password2)
# The 2 passwords must match # The 2 passwords must match
if password != password2: if password != password2:
return RedirectResponse('/userbootstrap') redirect('/userbootstrap')
# Create the admin user # Create the admin user
db.create_user(username, password, None, True, False) db.create_user(username, password, None, True, False)
# Redirect to the main page # Redirect to the main page
return RedirectResponse('/') redirect('/')
# Requested via GET; show the user creation UI # Requested via GET; show the user creation UI
else: else:
return TemplateResponse('userbootstrap.html', return template.render('userbootstrap.html',
setupname=config['InstanceName']) setupname=config['InstanceName'])

View file

@ -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

View file

@ -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')

View file

@ -0,0 +1,2 @@
from .sessions import start, end, put, get, has, delete

View file

@ -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]

View file

@ -0,0 +1,2 @@
from .template import init, render

View file

@ -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')

View file

@ -7,7 +7,7 @@ from threading import Lock, Thread, Timer
from time import sleep from time import sleep
import jinja2 import jinja2
from matemat.webserver import MatematWebserver, pagelet_cron from matemat.webserver.cron import cron
lock: Lock = Lock() lock: Lock = Lock()
@ -15,7 +15,7 @@ cron1called: int = 0
cron2called: int = 0 cron2called: int = 0
@pagelet_cron(seconds=4) @cron(seconds=4)
def cron1(config: Dict[str, str], def cron1(config: Dict[str, str],
jinja_env: jinja2.Environment, jinja_env: jinja2.Environment,
logger: logging.Logger) -> None: logger: logging.Logger) -> None:
@ -24,7 +24,7 @@ def cron1(config: Dict[str, str],
cron1called += 1 cron1called += 1
@pagelet_cron(seconds=3) @cron(seconds=3)
def cron2(config: Dict[str, str], def cron2(config: Dict[str, str],
jinja_env: jinja2.Environment, jinja_env: jinja2.Environment,
logger: logging.Logger) -> None: logger: logging.Logger) -> None:

View file

@ -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())

View file

@ -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')

View file

@ -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())

View file

@ -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'])

View file

@ -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

View file

@ -19,6 +19,7 @@ soda machine's touch screen).
packages=find_packages(exclude=['*.test']), packages=find_packages(exclude=['*.test']),
python_requires='>=3.6', python_requires='>=3.6',
install_requires=[ install_requires=[
'bottle',
'file-magic', 'file-magic',
'jinja2', 'jinja2',
'Pillow' 'Pillow'

View file

@ -70,7 +70,7 @@
<input type="submit" value="Save changes" /> <input type="submit" value="Save changes" />
</form> </form>
<script src="/js/touchkey.js" ></script> <script src="/static/js/touchkey.js" ></script>
<script> <script>
initTouchkey(true, 'touchkey-svg', null, 'admin-touchkey-touchkey'); initTouchkey(true, 'touchkey-svg', null, 'admin-touchkey-touchkey');
</script> </script>

View file

@ -5,7 +5,7 @@
{% block head %} {% block head %}
{# Show the setup name, as set in the config file, as tab title. Don't escape HTML entities. #} {# Show the setup name, as set in the config file, as tab title. Don't escape HTML entities. #}
<title>{{ setupname|safe }}</title> <title>{{ setupname|safe }}</title>
<link rel="stylesheet" href="/css/matemat.css"/> <link rel="stylesheet" href="/static/css/matemat.css"/>
{% endblock %} {% endblock %}
</head> </head>

View file

@ -26,7 +26,7 @@
</form> </form>
<a href="/">Cancel</a> <a href="/">Cancel</a>
<script src="/js/touchkey.js"></script> <script src="/static/js/touchkey.js"></script>
<script> <script>
initTouchkey(false, 'touchkey-svg', 'loginform', 'loginform-touchkey-value'); initTouchkey(false, 'touchkey-svg', 'loginform', 'loginform-touchkey-value');
</script> </script>