diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/chat/store.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/chat/store.py | 186 |
1 files changed, 153 insertions, 33 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/chat/store.py b/packages/meshbay-node/src/meshbay_node/chat/store.py index 6f23905..9e5ee90 100644 --- a/packages/meshbay-node/src/meshbay_node/chat/store.py +++ b/packages/meshbay-node/src/meshbay_node/chat/store.py @@ -1,9 +1,26 @@ """ MeshBay Node — SQLite-backed chat message store. -One database per group. Stores encrypted Sender Keys messages for offline -retrieval and history. Messages are stored as received (ciphertext) — -decryption happens on the client side. +One database per group. The node is a relay and an archive: it stores what it +was handed, serves it back, and — once a group has chat encryption switched on — +cannot read any of it. Decryption happens in the client, which is the only place +that holds the epoch key (`docs/chat-sender-keys.md` §5). + +Three things about the schema are load-bearing rather than incidental: + +* **`payload` is bytes, always.** It used to be a UTF-8 string in practice, and + the history path decoded it with `errors="replace"` — which substitutes + U+FFFD for every byte that is not valid UTF-8, i.e. for most of a ciphertext. + That would have corrupted history while live messages worked, which reads as + an intermittent decryption bug rather than as a wire-format error. +* **`format` says how to read a row**, so messages written before a group turned + encryption on keep rendering. Nothing is ever rewritten in place by the + switch; see `chat encrypt-history` for the explicit, backed-up alternative. +* **`(device, nonce)` is unique.** The nonce is 96 random bits chosen per + message by the sending device, so it is already required to be unique for + AES-GCM to be safe — making it a key costs nothing and turns a replayed + message (which is validly signed, being a copy of a real one) into an + integrity error instead of a duplicate. """ import logging @@ -15,6 +32,17 @@ import aiosqlite log = logging.getLogger(__name__) + +class ReplayedMessage(Exception): + """This device has already sent a message under this nonce.""" + +# Plaintext, as every message was before chat encryption existed. Rows keep it +# for ever; nothing rewrites them. +FORMAT_PLAIN = 0 +# Sealed under a chat epoch key: `payload` is the AES-256-GCM ciphertext, +# `nonce` its 96-bit nonce, `sig` the sender device's Ed25519 signature. +FORMAT_SEALED_V1 = 1 + _SCHEMA = """ CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -29,8 +57,27 @@ CREATE INDEX IF NOT EXISTS idx_messages_ts ON messages(timestamp); CREATE INDEX IF NOT EXISTS idx_messages_thread ON messages(thread_id); """ -_MIGRATE_SENDER_NAME = ( - "ALTER TABLE messages ADD COLUMN sender_name TEXT DEFAULT ''" +# Additive, every one with a default, so an existing chat.db opens unchanged and +# a node that is downgraded still reads its own rows. `CREATE TABLE IF NOT +# EXISTS` adds no column to a table that already exists — the same trap +# `create_all()` is recorded for on the hub — so each of these runs on its own +# and a duplicate-column error is the expected outcome on the second start. +_MIGRATIONS = ( + "ALTER TABLE messages ADD COLUMN sender_name TEXT DEFAULT ''", + "ALTER TABLE messages ADD COLUMN format INTEGER NOT NULL DEFAULT 0", + "ALTER TABLE messages ADD COLUMN epoch INTEGER NOT NULL DEFAULT 0", + "ALTER TABLE messages ADD COLUMN device BLOB DEFAULT NULL", + "ALTER TABLE messages ADD COLUMN nonce BLOB DEFAULT NULL", + "ALTER TABLE messages ADD COLUMN sig BLOB DEFAULT NULL", +) + +# A replay is a validly signed copy of a real message, so nothing about the +# signature refuses it. The nonce does: it is per message, per device, and a +# repeat is either an attack or a bug. Partial, because plaintext rows carry no +# nonce at all and NULLs are distinct in SQLite anyway. +_REPLAY_INDEX = ( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_replay " + "ON messages(device, nonce) WHERE nonce IS NOT NULL" ) @@ -43,11 +90,24 @@ class StoredMessage: timestamp: float thread_id: str | None sender_name: str = "" + format: int = FORMAT_PLAIN + epoch: int = 0 + device: bytes | None = None + nonce: bytes | None = None + sig: bytes | None = None + + +_COLUMNS = ("id, sender_id, iteration, payload, timestamp, thread_id, " + "sender_name, format, epoch, device, nonce, sig") def _row(r) -> StoredMessage: - return StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], payload=r[3], - timestamp=r[4], thread_id=r[5], sender_name=r[6] or "") + return StoredMessage( + id=r[0], sender_id=r[1], iteration=r[2], payload=r[3], + timestamp=r[4], thread_id=r[5], sender_name=r[6] or "", + format=r[7] or FORMAT_PLAIN, epoch=r[8] or 0, + device=r[9], nonce=r[10], sig=r[11], + ) class ChatStore: @@ -61,10 +121,17 @@ class ChatStore: self._db_path.parent.mkdir(parents=True, exist_ok=True) self._db = await aiosqlite.connect(str(self._db_path)) await self._db.executescript(_SCHEMA) - try: - await self._db.execute(_MIGRATE_SENDER_NAME) - except Exception: - pass + for statement in _MIGRATIONS: + try: + await self._db.execute(statement) + except Exception: + # Already applied. Swallowed per column rather than per batch: + # one loop with a shared `try` would stop at the first + # already-present column and silently skip every later one, so a + # node upgraded twice would be missing the newest fields with + # nothing to show for it. + pass + await self._db.execute(_REPLAY_INDEX) await self._db.commit() async def close(self) -> None: @@ -86,14 +153,32 @@ class ChatStore: payload: bytes, thread_id: str | None = None, sender_name: str = "", + *, + format: int = FORMAT_PLAIN, + epoch: int = 0, + device: bytes | None = None, + nonce: bytes | None = None, + sig: bytes | None = None, ) -> int: - """Store a message. Returns the row id.""" + """ + Store a message. Returns the row id. + + Raises `ReplayedMessage` if this device has already used this nonce — + see `_REPLAY_INDEX`. The caller must not turn that into a generic + failure the sender retries: it means the message is already stored. + """ ts = time.time() - cursor = await self._db.execute( - "INSERT INTO messages (sender_id, iteration, payload, timestamp, thread_id, sender_name) " - "VALUES (?, ?, ?, ?, ?, ?)", - (sender_id, iteration, payload, ts, thread_id, sender_name), - ) + try: + cursor = await self._db.execute( + "INSERT INTO messages (sender_id, iteration, payload, timestamp, " + " thread_id, sender_name, format, epoch, device, nonce, sig) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (sender_id, iteration, payload, ts, thread_id, sender_name, + format, epoch, device, nonce, sig), + ) + except aiosqlite.IntegrityError as e: + await self._db.rollback() + raise ReplayedMessage(str(e)) from e await self._db.commit() return cursor.lastrowid @@ -104,17 +189,12 @@ class ChatStore: ) -> list[StoredMessage]: """Get messages after a timestamp, most recent last.""" cursor = await self._db.execute( - "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name " + f"SELECT {_COLUMNS} " "FROM messages WHERE timestamp > ? ORDER BY timestamp ASC LIMIT ?", (since, limit), ) rows = await cursor.fetchall() - return [ - StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], - payload=r[3], timestamp=r[4], thread_id=r[5], - sender_name=r[6] or "") - for r in rows - ] + return [_row(r) for r in rows] async def get_recent(self, limit: int = 100) -> list[StoredMessage]: """The newest `limit` messages, oldest first so they render in order. @@ -125,7 +205,7 @@ class ChatStore: rows means ordering DESC in SQL and reversing here. """ cursor = await self._db.execute( - "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name " + f"SELECT {_COLUMNS} " "FROM messages ORDER BY id DESC LIMIT ?", (limit,), ) @@ -141,7 +221,7 @@ class ChatStore: `id` is AUTOINCREMENT: unique, and ordered by insertion. """ cursor = await self._db.execute( - "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name " + f"SELECT {_COLUMNS} " "FROM messages WHERE id < ? ORDER BY id DESC LIMIT ?", (before_id, limit), ) @@ -157,17 +237,57 @@ class ChatStore: async def get_thread(self, thread_id: str, limit: int = 100) -> list[StoredMessage]: """Get messages in a thread.""" cursor = await self._db.execute( - "SELECT id, sender_id, iteration, payload, timestamp, thread_id, sender_name " + f"SELECT {_COLUMNS} " "FROM messages WHERE thread_id = ? ORDER BY timestamp ASC LIMIT ?", (thread_id, limit), ) rows = await cursor.fetchall() - return [ - StoredMessage(id=r[0], sender_id=r[1], iteration=r[2], - payload=r[3], timestamp=r[4], thread_id=r[5], - sender_name=r[6] or "") - for r in rows - ] + return [_row(r) for r in rows] + + @property + def db_path(self) -> Path: + """Where this store lives — the re-encryption command backs it up.""" + return self._db_path + + async def count_by_format(self) -> tuple[int, int]: + """(plaintext, sealed). What the operator is deciding from.""" + cursor = await self._db.execute( + "SELECT format, COUNT(*) FROM messages GROUP BY format") + counts = {row[0]: row[1] for row in await cursor.fetchall()} + return (counts.get(FORMAT_PLAIN, 0), counts.get(FORMAT_SEALED_V1, 0)) + + async def all_plaintext(self) -> list[StoredMessage]: + """Every message still stored in the clear, oldest first.""" + cursor = await self._db.execute( + f"SELECT {_COLUMNS} FROM messages WHERE format = ? ORDER BY id", + (FORMAT_PLAIN,)) + return [_row(r) for r in await cursor.fetchall()] + + async def reseal(self, message_id: int, *, epoch: int, device: bytes, + nonce: bytes, ct: bytes, sig: bytes) -> None: + """ + Replace one plaintext row with its sealed form. **Does not commit** — + the caller commits once, so a re-encryption that fails half way leaves + the database as it was rather than half readable. + + `sender_name` is cleared because it moves inside the envelope; leaving + it would keep in the clear the one field the sealing was for. + """ + await self._db.execute( + "UPDATE messages SET format = ?, epoch = ?, device = ?, " + " nonce = ?, payload = ?, sig = ?, sender_name = '' " + "WHERE id = ?", + (FORMAT_SEALED_V1, epoch, device, nonce, ct, sig, message_id)) + + async def commit(self) -> None: + await self._db.commit() + + async def delete_older_than(self, cutoff: float) -> int: + """Retention. Returns how many rows went.""" + cursor = await self._db.execute( + "DELETE FROM messages WHERE timestamp < ?", (cutoff,)) + await self._db.commit() + return cursor.rowcount async def message_count(self) -> int: cursor = await self._db.execute("SELECT COUNT(*) FROM messages") |