From 0bfd4efab9016bc2f43e328a381f12a12d684b0f Mon Sep 17 00:00:00 2001 From: s3lph Date: Tue, 19 Jun 2018 21:34:48 +0200 Subject: [PATCH] Added unit tests for pagelet and static resource serving. --- matemat/webserver/httpd.py | 12 +- matemat/webserver/test/abstract_httpd_test.py | 11 +- matemat/webserver/test/test_serve.py | 114 ++++++++++++++++++ 3 files changed, 131 insertions(+), 6 deletions(-) create mode 100644 matemat/webserver/test/test_serve.py diff --git a/matemat/webserver/httpd.py b/matemat/webserver/httpd.py index 5c75520..a22969e 100644 --- a/matemat/webserver/httpd.py +++ b/matemat/webserver/httpd.py @@ -227,8 +227,13 @@ class HttpHandler(BaseHTTPRequestHandler): # 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): # Open and read the file - with open(filepath, 'rb') as f: - data = f.read() + try: + with open(filepath, 'rb') as f: + data = f.read() + except PermissionError: + self.send_error(403) + self.end_headers() + return # File read successfully, send 'OK' header self.send_response(200) # TODO: Guess the MIME type. Unfortunately this call solely relies on the file extension, not ideal? @@ -236,8 +241,9 @@ class HttpHandler(BaseHTTPRequestHandler): # Fall back to octet-stream type, if unknown if mimetype is None: mimetype = 'application/octet-stream' - # Send content type header + # Send content type and length header self.send_header('Content-Type', mimetype) + self.send_header('Content-Length', len(data)) self.end_headers() # Send the requested resource as response body self.wfile.write(data) diff --git a/matemat/webserver/test/abstract_httpd_test.py b/matemat/webserver/test/abstract_httpd_test.py index eda18dd..de0daf6 100644 --- a/matemat/webserver/test/abstract_httpd_test.py +++ b/matemat/webserver/test/abstract_httpd_test.py @@ -3,6 +3,7 @@ from typing import Any, Callable, Dict, Tuple, Union import unittest.mock from io import BytesIO +from tempfile import TemporaryDirectory from abc import ABC from datetime import datetime @@ -45,6 +46,8 @@ class HttpResponse: """ Parse a new fragment of data. This function does nothing if the parsed HTTP response is already complete. + DO NOT USE THIS OUTSIDE UNIT TESTING! + :param fragment: The data fragment to parse. """ # response packet complete, nothing to do @@ -178,10 +181,12 @@ class AbstractHttpdTest(ABC, unittest.TestCase): self.client_sock.set_request(b'GET /just/testing/sessions HTTP/1.1\r\n\r\n') handler = HttpHandler(self.client_sock, ('::1', 45678), self.server) packet = self.client_sock.get_response() - - TODO(s3lph): This could probably go here instead. """ def setUp(self) -> None: - self.server: HTTPServer = MockServer() + self.tempdir: TemporaryDirectory = TemporaryDirectory(prefix='matemat.', dir='/tmp/') + self.server: HTTPServer = MockServer(webroot=self.tempdir.name) self.client_sock: MockSocket = MockSocket() + + def tearDown(self): + self.tempdir.cleanup() diff --git a/matemat/webserver/test/test_serve.py b/matemat/webserver/test/test_serve.py new file mode 100644 index 0000000..0c59a0c --- /dev/null +++ b/matemat/webserver/test/test_serve.py @@ -0,0 +1,114 @@ + +from typing import Any, Dict + +import os +import os.path +from matemat.webserver.httpd import HttpHandler +from matemat.webserver.test.abstract_httpd_test import AbstractHttpdTest, test_pagelet + + +@test_pagelet('/just/testing/serve_pagelet_ok') +def serve_test_pagelet_ok(method: str, + path: str, + args: Dict[str, str], + session_vars: Dict[str, Any], + headers: Dict[str, str]): + headers['Content-Type'] = 'text/plain' + return 200, 'serve test pagelet ok' + + +@test_pagelet('/just/testing/serve_pagelet_fail') +def serve_test_pagelet_fail(method: str, + path: str, + args: Dict[str, str], + session_vars: Dict[str, Any], + headers: Dict[str, str]): + session_vars['test'] = 'hello, world!' + headers['Content-Type'] = 'text/plain' + return 500, 'serve test pagelet fail' + + +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) + + def test_serve_pagelet_ok(self): + # Call the test pagelet that produces a 200 OK result + self.client_sock.set_request(b'GET /just/testing/serve_pagelet_ok 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_ok', packet.pagelet) + # Make sure the expected content is served + self.assertEqual(200, packet.statuscode) + self.assertEqual('serve test pagelet ok', 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 the correct pagelet was called + self.assertEqual('serve_test_pagelet_fail', packet.pagelet) + # Make sure the expected content is served + self.assertEqual(500, packet.statuscode) + self.assertEqual('serve test pagelet fail', 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('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) + + 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)