1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
import http.client
import tempfile
import threading
import unittest
from pathlib import Path
from server import MAX_FILE_BYTES, detect_image_type, make_server
PNG = b"\x89PNG\r\n\x1a\n" + b"test-image"
def multipart(filename: str, payload: bytes, boundary: str = "test-boundary") -> tuple[bytes, str]:
body = (
f"--{boundary}\r\n"
f'Content-Disposition: form-data; name="image"; filename="{filename}"\r\n'
"Content-Type: application/octet-stream\r\n\r\n"
).encode() + payload + f"\r\n--{boundary}--\r\n".encode()
return body, f"multipart/form-data; boundary={boundary}"
class ServerTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.upload_dir = Path(self.temp.name) / "private"
self.server = make_server("127.0.0.1", 0, self.upload_dir)
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
self.thread.start()
self.port = self.server.server_address[1]
def tearDown(self):
self.server.shutdown()
self.server.server_close()
self.thread.join()
self.temp.cleanup()
def request(self, method, path, body=None, headers=None):
connection = http.client.HTTPConnection("127.0.0.1", self.port, timeout=3)
connection.request(method, path, body=body, headers=headers or {})
response = connection.getresponse()
data = response.read()
connection.close()
return response, data
def test_accepts_png_and_stores_it_under_target(self):
body, content_type = multipart("photo.png", PNG)
response, data = self.request(
"POST", "/logo-contest", body, {"Content-Type": content_type, "Content-Length": str(len(body))}
)
self.assertEqual(response.status, 201, data)
target_dir = self.upload_dir / "logo-contest"
files = list(target_dir.iterdir())
self.assertEqual(len(files), 1)
self.assertEqual(files[0].read_bytes(), PNG)
self.assertNotIn("photo", files[0].name)
self.assertEqual(files[0].stat().st_mode & 0o777, 0o600)
self.assertEqual(target_dir.stat().st_mode & 0o777, 0o700)
hidden_response, _ = self.request("GET", f"/logo-contest/{files[0].name}")
self.assertEqual(hidden_response.status, 404)
def test_root_is_rejected_but_target_page_is_available(self):
root_response, _ = self.request("GET", "/")
target_response, page = self.request("GET", "/911u19jasdl")
self.assertEqual(root_response.status, 418)
self.assertEqual(target_response.status, 200)
self.assertIn(b"Private Drop", page)
def test_upload_to_root_is_rejected_as_teapot(self):
body, content_type = multipart("photo.png", PNG)
response, _ = self.request(
"POST", "/", body, {"Content-Type": content_type, "Content-Length": str(len(body))}
)
self.assertEqual(response.status, 418)
def test_targets_are_stored_in_separate_directories(self):
for target in ("logo-contest", "911u19jasdl"):
body, content_type = multipart("photo.png", PNG)
response, _ = self.request(
"POST", f"/{target}", body, {"Content-Type": content_type, "Content-Length": str(len(body))}
)
self.assertEqual(response.status, 201)
self.assertEqual(len(list((self.upload_dir / target).iterdir())), 1)
def test_invalid_or_nested_targets_are_rejected(self):
for path in ("/bad.target", "/nested/target", "/%2e%2e", "/healthz"):
body, content_type = multipart("photo.png", PNG)
response, _ = self.request(
"POST", path, body, {"Content-Type": content_type, "Content-Length": str(len(body))}
)
self.assertEqual(response.status, 404)
def test_rejects_non_image_content(self):
body, content_type = multipart("fake.jpg", b"not really an image")
response, _ = self.request(
"POST", "/logo-contest", body, {"Content-Type": content_type, "Content-Length": str(len(body))}
)
self.assertEqual(response.status, 400)
self.assertFalse((self.upload_dir / "logo-contest").exists())
def test_rejects_oversized_request_before_reading_body(self):
response, _ = self.request(
"POST", "/logo-contest", b"", {"Content-Type": "multipart/form-data; boundary=x", "Content-Length": str(MAX_FILE_BYTES + 1024 * 1024 + 1)}
)
self.assertEqual(response.status, 413)
def test_unknown_paths_are_not_served(self):
response, _ = self.request("GET", "/server.py")
self.assertEqual(response.status, 404)
def test_detects_supported_signatures(self):
self.assertEqual(detect_image_type(b"\xff\xd8\xffrest"), (".jpg", "image/jpeg"))
self.assertEqual(detect_image_type(PNG), (".png", "image/png"))
self.assertEqual(detect_image_type(b"\x00\x00\x00\x18ftypheic\x00\x00\x00\x00heic"), (".heic", "image/heic"))
if __name__ == "__main__":
unittest.main()
|
