#!/usr/bin/env python3 """server.py: file allowlist, headers, blob checks, atomic saves, backups. python3 tests/test_server.py """ import base64 import datetime import http.client import json import os import sys import tempfile import threading import unittest from http.server import ThreadingHTTPServer sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import server # noqa: E402 def blob(ct_len=40): b = lambda n: base64.b64encode(os.urandom(n)).decode() return json.dumps({"v": 1, "salt": b(16), "iv": b(12), "ct": b(ct_len)}).encode() class ServerTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() server.Handler.store = server.Store(self.tmp.name, 0, 3, 2) self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), server.Handler) threading.Thread(target=self.httpd.serve_forever, daemon=True).start() self.port = self.httpd.server_address[1] def tearDown(self): self.httpd.shutdown() self.httpd.server_close() self.tmp.cleanup() def req(self, method, path, body=None, headers=None): c = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5) h = {"Content-Type": "application/json"} if body is not None else {} h.update(headers or {}) c.request(method, path, body=body, headers=h) r = c.getresponse() data = r.read() c.close() return r, data def test_health_and_headers(self): r, data = self.req("GET", "/api/health") self.assertEqual((r.status, data), (200, b'{"ok":true}')) csp = r.getheader("Content-Security-Policy") self.assertIn("default-src 'none'", csp) self.assertIn("frame-ancestors 'none'", csp) self.assertEqual(r.getheader("X-Content-Type-Options"), "nosniff") self.assertEqual(r.getheader("X-Frame-Options"), "DENY") self.assertEqual(r.getheader("Referrer-Policy"), "no-referrer") def test_serves_only_app_files(self): for path in ("/", "/index.html", "/app.js", "/theme.js", "/styles.css", "/crypto-fallback.js", "/banks/monzo.svg", "/vendor/noble-hashes-2.2.0/pbkdf2.js", "/vendor/noble-ciphers-2.2.0/aes.js"): r, _ = self.req("GET", path) self.assertEqual(r.status, 200, path) r, _ = self.req("GET", "/app.js") self.assertTrue(r.getheader("Content-Type").startswith("text/javascript")) for path in ("/server.py", "/Dockerfile", "/README.md", "/LICENSE", "/data/", "/data/ledger.enc", "/banks/", "/banks/NOTICE.txt", "/vendor/", "/vendor/README.md", "/vendor/noble-hashes-2.2.0/LICENSE", "/vendor/noble-hashes-2.2.0/argon2.js", "/tests/test_server.py", "/../server.py", "/%2e%2e/server.py", "/banks/..%2fserver.py", "/.git/config", "//etc/passwd", "/app.js.map"): r, _ = self.req("GET", path) self.assertEqual(r.status, 404, path) def test_round_trip_and_validation(self): r, _ = self.req("GET", "/api/data") self.assertEqual(r.status, 204) good = blob() r, _ = self.req("PUT", "/api/data", good) self.assertEqual(r.status, 200) r, data = self.req("GET", "/api/data") self.assertEqual((r.status, data), (200, good)) self.assertEqual(os.stat(os.path.join(self.tmp.name, "ledger.enc")).st_mode & 0o777, 0o600) bad = [b"not json", b"[]", b'{"salt":"a","iv":"b"}', json.dumps({"salt": "!!", "iv": "AAAAAAAAAAAAAAAA", "ct": "AAAAAAAAAAAAAAAAAAAAAA=="}).encode(), json.dumps({**json.loads(blob()), "extra": 1}).encode(), json.dumps({**json.loads(blob()), "v": "1"}).encode(), json.dumps({**json.loads(blob()), "salt": base64.b64encode(b"short").decode()}).encode(), blob(ct_len=4)] for body in bad: r, _ = self.req("PUT", "/api/data", body) self.assertEqual(r.status, 400, body) r, _ = self.req("PUT", "/api/data", good, {"Content-Type": "text/plain"}) self.assertEqual(r.status, 415) r, _ = self.req("PUT", "/api/data", good, {"Origin": "http://evil.example"}) self.assertEqual(r.status, 403) r, _ = self.req("PUT", "/api/data", good, {"Sec-Fetch-Site": "cross-site"}) self.assertEqual(r.status, 403) r, _ = self.req("PUT", "/api/data", good, {"Origin": f"http://127.0.0.1:{self.port}", "Sec-Fetch-Site": "same-origin"}) self.assertEqual(r.status, 200) r, _ = self.req("POST", "/api/data", good) self.assertEqual(r.status, 405) r, _ = self.req("PUT", "/index.html", good) self.assertEqual(r.status, 405) r, data = self.req("GET", "/api/data") self.assertEqual(data, good) # none of the rejected bodies landed def test_backups_rotate(self): store = server.Handler.store bodies = [blob() for _ in range(7)] for b in bodies: r, _ = self.req("PUT", "/api/data", b) self.assertEqual(r.status, 200) snaps = store._snapshots() self.assertEqual(len(snaps), 3) # keep_recent=3, all taken today with open(os.path.join(store.backups, snaps[-1]), "rb") as f: self.assertEqual(f.read(), bodies[-2]) # the version before the last save # Older days: the last snapshot of each of the 2 newest days survives. for day in ("20260101", "20260102", "20260103"): for t in ("T080000000000Z", "T200000000000Z"): with open(os.path.join(store.backups, f"ledger-{day}{t}.enc"), "wb") as f: f.write(b"old") self.req("PUT", "/api/data", blob()) snaps = store._snapshots() today = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d") self.assertNotIn("ledger-20260101T200000000000Z.enc", snaps) self.assertNotIn("ledger-20260102T200000000000Z.enc", snaps) self.assertNotIn("ledger-20260103T080000000000Z.enc", snaps) self.assertIn("ledger-20260103T200000000000Z.enc", snaps) self.assertEqual(len([s for s in snaps if s[7:15] == today]), 3) self.assertEqual(len(snaps), 4) self.assertEqual([n for n in os.listdir(store.backups) if n.endswith(".tmp")], []) def test_backup_interval(self): server.Handler.store = store = server.Store(self.tmp.name, 3600, 10, 30) for _ in range(4): self.req("PUT", "/api/data", blob()) self.assertEqual(len(store._snapshots()), 1) if __name__ == "__main__": unittest.main(verbosity=2)