Start LibreLedger: an encrypted money ledger that stores only ciphertext
LibreLedger is a single-person money ledger. The browser encrypts
everything (PBKDF2-HMAC-SHA256 with 250k iterations, then AES-256-GCM),
and a small standard-library Python server stores only the resulting
{v, salt, iv, ct} blob. Ledger files from the earlier Money Ledger version
open unchanged.
- On plain http to a LAN or VPN address, where browsers hide Web Crypto,
the app switches to vendored @noble/hashes and @noble/ciphers 2.2.0 and
says so on the lock screen. tests/crypto-interop.test.mjs shows both
paths read each other's files.
- The server hands out only the app's own files, sends a self-only CSP,
nosniff, frame denial and no-referrer, checks the shape of each blob,
refuses cross-site writes and writes atomically. It keeps the last 10
backups plus one per day for 30 days.
- The container runs that server from a digest-pinned python alpine
image, as an unprivileged user, with its data in /data and a health
check on /api/health.
Assisted-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,144 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user