Serverjonas/crushdb
SHA256
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rewirite to new name

Browse Source
This commit is contained in:
jhkaktus
2026-10-14 20:29:22 +02:00
parent 2120c56f45
commit 1cdb3fa784
23 changed files with 2521 additions and 5755 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/test_integrity.py
+272 -82
View File
@@ -1,101 +1,291 @@
"""Integrity tests: tampering and replay of old records must be detected.
import os
import sys
import shutil
Server-side guarantees tested here:
- Updating with a stale version (replay of an old ciphertext) -> 409
- Deleting with a stale version -> 409
- Duplicate uuid creation -> 409
- Ciphertext tampering is impossible server-side without breaking the
version check chain: any modification requires pushing a NEW version,
which the client detects via its own version + AEAD tag on decrypt.
# Use an isolated DB per test run
TEST_DB = "instance/test_integrity.sqlite"
Client-side guarantees (AEAD/AAD binding; exercised via crypto.js):
- A ciphertext swapped between records fails the Poly1305 tag.
- A modified salt breaks the tag (re-key attempt detected).
- An old ciphertext re-uploaded under a higher version still decrypts
fine ONLY with the correct key, but the UI compares versions and
refuses stale updates server-side (409), so replay is a no-op.
os.environ["WHO2_DB_PATH"] = TEST_DB
Run: ./venv/bin/python -m pytest backend/test_integrity.py
"""
import base64
import pytest
from backend.app import app
from backend import storage
import backend.storage as storage
@pytest.fixture()
def client(tmp_path, monkeypatch):
monkeypatch.setattr(storage, "DB_PATH", str(tmp_path / "test.sqlite3"))
def _clean():
if os.path.exists(TEST_DB):
os.remove(TEST_DB)
if os.path.exists(TEST_DB + "-journal"):
os.remove(TEST_DB + "-journal")
if os.path.exists(TEST_DB + "-wal"):
os.remove(TEST_DB + "-wal")
if os.path.exists(TEST_DB + "-shm"):
os.remove(TEST_DB + "-shm")
def _reset():
_clean()
storage.init_db()
with app.test_client() as c:
yield c
def b64(data: bytes) -> str:
return base64.b64encode(data).decode()
def test_create_and_authenticate():
_reset()
email = "alice@example.com"
password = "Sup3rS3cret"
acc = storage.create_account(email, password)
assert acc["email"] == email
# first account becomes the bootstrap admin
assert acc["is_admin"] is True
assert storage.authenticate(email, password) is not None
assert storage.authenticate(email, "WRONG") is None
assert storage.authenticate("noone@example.com", password) is None
def make_body(uuid: str, marker: bytes):
return {
"uuid": uuid,
"ciphertext": b64(marker),
"nonce": b64(b"\x01" * 24),
"salt": b64(b"\x02" * 16),
}
def test_only_admins_create_persons():
_reset()
admin = storage.create_account("admin@x.com", "pw")
user = storage.create_account("user@x.com", "pw")
assert storage.is_admin(admin)
assert not storage.is_admin(user)
try:
storage.create_person(user, "Nope")
assert False, "expected PermissionError"
except PermissionError:
pass
# admin creates a person and gives it the user's email address
person = storage.create_person_with_account(admin, "User Person", "user@x.com")
refreshed = storage.get_account_by_id(user["id"])
assert refreshed["person_id"] == person["id"]
assert person["emails"] == ["user@x.com"]
# a non-admin cannot link or unlink emails either
try:
storage.attach_email_to_person(user, person["id"], "admin@x.com")
assert False, "expected PermissionError"
except PermissionError:
pass
def test_stale_update_is_refused(client):
uuid = "replay-test"
old_body = make_body(uuid, b"OLD-CIPHERTEXT-v1")
def test_person_account_linking():
_reset()
admin = storage.create_account("admin@x.com", "pw")
# separate account so linking can be exercised explicitly
a = storage.create_account("a@x.com", "pw")
p = storage.create_person(admin, "Alice")
storage.link_account_to_person(admin, p["id"], a["id"])
person = storage.get_person(p["id"])
assert person["emails"] == ["a@x.com"]
r = client.post("/api/records", json=old_body)
assert r.status_code == 201
v1 = r.get_json()
assert v1["version"] == 1
# one account can only ever be linked to one person
p2 = storage.create_person(admin, "Bob")
try:
storage.attach_email_to_person(admin, p2["id"], "a@x.com")
assert False, "expected ValueError"
except ValueError:
pass
# legitimate update pushes version to 2
new_body = make_body(uuid, b"NEW-CIPHERTEXT-v2")
r = client.put(f"/api/records/{uuid}", json={**new_body, "version": 1})
assert r.status_code == 200
assert r.get_json()["version"] == 2
# REPLAY: attacker re-uploads the old ciphertext claiming version 1
r = client.put(f"/api/records/{uuid}", json={**old_body, "version": 1})
assert r.status_code == 409
# stale version 1 also cannot delete the record
r = client.delete(f"/api/records/{uuid}", json={"version": 1})
assert r.status_code == 409
# record still holds v2 content
rec = storage.get_record(uuid)
assert rec["ciphertext"] == b"NEW-CIPHERTEXT-v2"
assert rec["version"] == 2
storage.attach_email_to_person(admin, p2["id"], "admin@x.com")
assert "admin@x.com" in storage.get_person(p2["id"])["emails"]
# detaching works and is admin-only
storage.detach_email_from_person(admin, p2["id"], "admin@x.com")
assert storage.get_person(p2["id"])["emails"] == []
try:
storage.detach_email_from_person(a, p["id"], "a@x.com")
assert False, "expected PermissionError"
except PermissionError:
pass
def test_duplicate_uuid_refused(client):
body = make_body("dup-test", b"X")
assert client.post("/api/records", json=body).status_code == 201
assert client.post("/api/records", json=body).status_code == 409
def test_article_visibility():
_reset()
owner = storage.create_account("owner@x.com", "pw")
storage.bootstrap_admin(owner["id"])
owner["rights"] = list(storage.RIGHTS.keys())
admin_person = storage.create_person(owner, "Owner")
owner["person_id"] = admin_person["id"]
other = storage.create_account("other@x.com", "pw")
storage.grant_right(owner, other["id"], "view_private")
other["person_id"] = storage.create_person_with_account(owner, "Other", "other@x.com")["id"]
other["rights"] = ["view_private"]
storage.create_article(
owner, owner["person_id"], "notes", "Secret",
"hidden content", visibility="private",
)
articles = storage.list_articles(other, owner["person_id"])
assert len(articles) == 0 or articles[0].get("content") == ""
storage.grant_right(owner, owner["id"], "view_private")
articles = storage.list_articles(owner, owner["person_id"])
assert any(a["title"] == "Secret" and a.get("content") == "hidden content" for a in articles)
def test_delete_then_recreate_replay_still_safe(client):
# Deleting a record frees the uuid; recreate starts again at v1.
# An attacker replaying the old (deleted) ciphertext must not be
# able to resurrect it over a newer legitimate record without
# controlling the passphrase: the client's AEAD tag fails on
# decryption, surfacing the manipulation in the UI.
body = make_body("res-test", b"DELETED-RECORD")
client.post("/api/records", json=body)
assert client.delete("/api/records/res-test", json={"version": 1}).status_code == 204
def test_group_based_visibility():
_reset()
owner = storage.create_account("owner@x.com", "pw")
storage.bootstrap_admin(owner["id"])
owner["rights"] = list(storage.RIGHTS.keys())
owner["person_id"] = storage.create_person(owner, "Owner")["id"]
reader = storage.create_account("reader@x.com", "pw")
storage.grant_right(owner, reader["id"], "view_private")
reader["person_id"] = storage.create_person_with_account(owner, "Reader", "reader@x.com")["id"]
reader["rights"] = ["view_private"]
# recreate by attacker: uuid free again -> succeeds tamper-wise
r = client.post("/api/records", json=body)
assert r.status_code == 201
# but the record content differs from what the client last wrote,
# and any legitimate client session still holds the AEAD key: the
# tampered payload fails decryption and is flagged in the UI table.
assert storage.get_record("res-test")["ciphertext"] == b"DELETED-RECORD"
group = storage.create_group(owner, owner["person_id"], "Team")
storage.add_group_member(owner, group["id"], reader["person_id"])
storage.create_article(
owner, owner["person_id"], "notes", "Shared",
"shared content", visibility="group",
group_ids=[group["id"]],
)
noaccess = storage.list_articles(None, owner["person_id"])
assert not any(a["title"] == "Shared" and a.get("content") == "shared content" for a in noaccess)
view = storage.list_articles(reader, owner["person_id"])
assert any(a["title"] == "Shared" and a.get("content") == "shared content" for a in view)
def test_normal_user_only_notes_and_own_groups():
_reset()
admin = storage.create_account("admin@x.com", "pw")
me = storage.create_person_with_account(admin, "Me", "me@x.com", "pw")
friend = storage.create_person_with_account(admin, "Friend", "friend@x.com", "pw")
user = storage.authenticate("me@x.com", "pw")
assert user["person_id"] == me["id"]
assert not user["is_admin"]
# a normal user may add notes to their own person only
storage.create_article(user, me["id"], "notes", "Note", "hello", visibility="private")
try:
storage.create_article(user, me["id"], "links", "Link", "http://x")
assert False, "expected PermissionError"
except PermissionError:
pass
try:
storage.create_article(user, friend["id"], "notes", "Peek", "nope")
assert False, "expected PermissionError"
except PermissionError:
pass
# own group + invite a friend into it
g = storage.create_group(user, me["id"], "My Team")
storage.add_group_member(user, g["id"], friend["id"])
assert any(m["id"] == friend["id"] for m in storage.get_group(g["id"])["members"])
friend_user = storage.authenticate("friend@x.com", "pw")
admin_person = storage.create_person(admin, "Admin")
foreign = storage.create_group(admin, admin_person["id"], "Admins")
# the friend is a member, so they see the group but not foreign groups
assert {x["id"] for x in storage.list_groups(friend_user)} == {g["id"]}
assert foreign["id"] not in {x["id"] for x in storage.list_groups(friend_user)}
assert not storage.group_is_visible_to(friend_user, foreign["id"])
# a member may invite, a bystander may not
try:
storage.add_group_member(friend_user, foreign["id"], friend_user["person_id"])
assert False, "expected PermissionError"
except PermissionError:
pass
# only admins may read the rights list
try:
storage.list_rights(user)
assert False, "expected PermissionError"
except PermissionError:
pass
def test_ensure_admin_promotes_oldest_account():
_reset()
storage.create_account("first@x.com", "pw")
storage.create_account("second@x.com", "pw")
conn = storage._conn()
conn.execute("DELETE FROM account_rights") # simulate a database from before roles existed
conn.commit()
conn.close()
assert storage.authenticate("first@x.com", "pw")["is_admin"] is False
storage.ensure_admin()
assert storage.authenticate("first@x.com", "pw")["is_admin"] is True
assert storage.authenticate("second@x.com", "pw")["is_admin"] is False
def test_article_type_constraint():
_reset()
acc = storage.create_account("x@x.com", "pw")
acc["person_id"] = storage.create_person(acc, "X")["id"]
try:
storage.create_article(acc, acc["person_id"], "bogus", "T", "C")
assert False, "expected ValueError"
except ValueError as e:
assert "Invalid article type" in str(e)
def test_right_management():
_reset()
admin = storage.create_account("admin@x.com", "pw")
storage.bootstrap_admin(admin["id"])
admin["rights"] = list(storage.RIGHTS.keys())
admin["person_id"] = storage.create_person(admin, "Admin")["id"]
storage.grant_right(admin, admin["id"], "view_private")
storage.grant_right(admin, admin["id"], "manage_articles")
rights = storage.list_rights(admin, admin["id"])
assert any(r["id"] == admin["id"] and "view_private" in r["rights"] for r in rights)
storage.revoke_right(admin, admin["id"], "manage_articles")
rights = storage.list_rights(admin)
assert any(r["id"] == admin["id"] and "view_private" in r["rights"] for r in rights)
assert not any(r["id"] == admin["id"] and "manage_articles" in r["rights"] for r in rights)
def test_delete_person_cascades():
_reset()
acc = storage.create_account("x@x.com", "pw")
acc["person_id"] = storage.create_person(acc, "X")["id"]
g = storage.create_group(acc, acc["person_id"], "G")
storage.add_group_member(acc, g["id"], acc["person_id"])
storage.create_article(acc, acc["person_id"], "notes", "T", "C")
ok = storage.delete_person(acc, acc["person_id"])
assert ok
assert storage.get_person(acc["person_id"]) is None
assert storage.get_group(g["id"]) is None
def run_all():
tests = [
test_create_and_authenticate,
test_only_admins_create_persons,
test_person_account_linking,
test_article_visibility,
test_group_based_visibility,
test_normal_user_only_notes_and_own_groups,
test_ensure_admin_promotes_oldest_account,
test_article_type_constraint,
test_right_management,
test_delete_person_cascades,
]
passed = 0
failed = 0
last_err = None
for t in tests:
try:
t()
passed += 1
print("PASS", t.__name__)
except AssertionError as e:
failed += 1
last_err = e
print("FAIL", t.__name__, "-", str(e))
except Exception as e:
failed += 1
last_err = e
print("ERROR", t.__name__, "-", str(e))
_clean()
if failed:
print(f"\n{passed} passed, {failed} failed")
raise SystemExit(1)
print(f"\nAll {passed} tests passed")
return 0
if __name__ == "__main__":
sys.exit(run_all())