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
119
120
121
122
123
124
|
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import json
import os
import socket
import tempfile
import threading
import unittest
import http.client
import deckit.module as mod
import deckit.server as srv
TOKEN = "test-token"
HOST = {"Host": "127.0.0.1:8765", "Cookie": f"deckit={TOKEN}"}
ORIGIN = {"Origin": "http://127.0.0.1:8765"}
class UnixConnection(http.client.HTTPConnection):
def __init__(self, path):
super().__init__("localhost")
self.path = path
def connect(self):
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.sock.connect(self.path)
class Server(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.dir = tempfile.TemporaryDirectory()
cls.sock = os.path.join(cls.dir.name, "http.sock")
srv.TOKEN = TOKEN
srv.REGISTRY = mod.Registry([
mod.Module("demo", "Demo", lambda: {"ok": True},
{"ping": lambda body: {"got": body.get("n")},
"boom": lambda body: int(body.get("x"))}),
])
cls.server = srv.UnixHTTPServer(cls.sock, srv.Handler)
threading.Thread(target=cls.server.serve_forever, daemon=True).start()
@classmethod
def tearDownClass(cls):
cls.server.shutdown()
cls.server.server_close()
cls.dir.cleanup()
def req(self, method, path, headers=None, body=None):
conn = UnixConnection(self.sock)
conn.request(method, path, body=body, headers=headers or {})
r = conn.getresponse()
data = r.read()
conn.close()
return r.status, data
def test_assets_and_api(self):
self.assertEqual(self.req("GET", "/", HOST)[0], 200)
self.assertEqual(self.req("GET", "/theme.css", HOST)[0], 200)
self.assertEqual(json.loads(self.req("GET", "/api/modules", HOST)[1]),
[{"name": "demo", "title": "Demo"}])
self.assertEqual(json.loads(self.req("GET", "/api/demo", HOST)[1]), {"ok": True})
def test_trust_and_auth(self):
self.assertEqual(self.req("GET", "/api/demo", {"Host": "evil.com"})[0], 403)
self.assertEqual(self.req("POST", "/api/demo/x", {**HOST, "Origin": "https://evil.com"})[0], 403)
self.assertEqual(self.req("GET", "/api/demo", {"Host": "127.0.0.1:8765"})[0], 401)
def test_unlock_and_bad_action(self):
conn = UnixConnection(self.sock)
conn.request("GET", f"/?t={TOKEN}", headers={"Host": "127.0.0.1:8765"})
r = conn.getresponse()
r.read()
conn.close()
self.assertEqual(r.status, 303)
self.assertIn("deckit=", r.getheader("Set-Cookie"))
self.assertIn("SameSite=Strict", r.getheader("Set-Cookie"))
self.assertEqual(self.req("POST", "/api/demo/nope", {**HOST, **ORIGIN})[0], 404)
def test_rejects_oversized_body(self):
big = "x" * 5000
status, _ = self.req("POST", "/api/demo/ping",
{**HOST, **ORIGIN, "Content-Type": "application/json"}, big)
self.assertEqual(status, 400)
def test_rejects_wrong_typed_body(self):
h = {**HOST, **ORIGIN, "Content-Type": "application/json"}
self.assertEqual(self.req("POST", "/api/demo/boom", h, '{"x": null}')[0], 400)
def test_icons_route_rejects_traversal(self):
self.assertEqual(self.req("GET", "/icons/..%2Fconfig.toml", HOST)[0], 404)
def test_icon_theme_follows_gsettings(self):
from unittest import mock
srv._theme_cache.update(t=0, v=None)
with mock.patch.object(srv, "_run", return_value="'some-theme'"):
self.assertEqual(srv.icon_theme(), "some-theme")
srv._theme_cache.update(t=0, v=None)
def test_icon_path_resolves_and_rejects(self):
import tempfile
from pathlib import Path
base = Path(tempfile.mkdtemp())
d = base / "candy-icons" / "apps" / "scalable"
d.mkdir(parents=True)
(d / "obs.svg").write_text("<svg/>")
self.assertEqual(srv.icon_path("obs", "candy-icons", [base]), d / "obs.svg")
self.assertIsNone(srv.icon_path("../x", "candy-icons", [base]))
self.assertIsNone(srv.icon_path("nope", "candy-icons", [base]))
def test_art_route(self):
from unittest import mock
with mock.patch.object(srv, "art", return_value=(b"PNGDATA", "image/png")):
status, body = self.req("GET", "/api/art", HOST)
self.assertEqual(status, 200)
self.assertEqual(body, b"PNGDATA")
if __name__ == "__main__":
unittest.main()
|
