sigil/sigil/sealed.py

"""Sealed store: encryption at rest for the Sigil database file.

Tokens, codes and bootstrap secrets are already stored as digests, so a stolen
cleartext SQLite file does not yield live credentials. It does still yield the
full agent roster, revocation state, and audit history. A sealed store wraps
the on-disk bytes in AES-256-GCM so that material is unreadable without the
passphrase.

This is an incremental path short of SQLCipher (no page-level encryption, the
working copy is plaintext in memory / a temp file while open). It is a real
working seal: the file on disk is ciphertext, and tests assert a stolen seal
has no SQLite header and no recoverable agent rows without the passphrase.

Format on disk (version 1):

    b"SIGILSEAL1" + 16-byte salt + 12-byte nonce + AESGCM(ciphertext)
    key = PBKDF2-HMAC-SHA256(passphrase, salt, 480_000 iterations, 32 bytes)
"""

from __future__ import annotations

import os
import sqlite3
import tempfile
from typing import Optional

from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives import hashes

from .store import Store

MAGIC = b"SIGILSEAL1"
SALT_LEN = 16
NONCE_LEN = 12
KEY_LEN = 32
PBKDF2_ITERS = 480_000
MIN_PASSPHRASE = 12


def derive_key(passphrase: str, salt: bytes) -> bytes:
    if len(passphrase) < MIN_PASSPHRASE:
        raise ValueError(
            f"seal passphrase must be at least {MIN_PASSPHRASE} characters"
        )
    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=KEY_LEN,
        salt=salt,
        iterations=PBKDF2_ITERS,
    )
    return kdf.derive(passphrase.encode("utf-8"))


def seal_bytes(plaintext: bytes, passphrase: str, *, salt: Optional[bytes] = None) -> bytes:
    """Encrypt plaintext DB bytes into a sealed blob."""
    salt = salt or os.urandom(SALT_LEN)
    key = derive_key(passphrase, salt)
    nonce = os.urandom(NONCE_LEN)
    ct = AESGCM(key).encrypt(nonce, plaintext, MAGIC)
    return MAGIC + salt + nonce + ct


def unseal_bytes(blob: bytes, passphrase: str) -> bytes:
    """Decrypt a sealed blob back to plaintext DB bytes."""
    if not blob.startswith(MAGIC):
        raise ValueError("not a Sigil sealed store (bad magic)")
    salt = blob[len(MAGIC):len(MAGIC) + SALT_LEN]
    nonce_start = len(MAGIC) + SALT_LEN
    nonce = blob[nonce_start:nonce_start + NONCE_LEN]
    ct = blob[nonce_start + NONCE_LEN:]
    key = derive_key(passphrase, salt)
    try:
        return AESGCM(key).decrypt(nonce, ct, MAGIC)
    except Exception as exc:
        raise ValueError("wrong passphrase or corrupted sealed store") from exc


def is_sealed_file(path: str) -> bool:
    if not os.path.exists(path):
        return False
    with open(path, "rb") as handle:
        return handle.read(len(MAGIC)) == MAGIC


class SealedStore(Store):
    """Store whose on-disk file is AES-GCM sealed.

    Open decrypts into a temp SQLite file (or creates an empty one). Call
    `flush()` or `close()` to re-encrypt and write the seal back to `path`.
    The passphrase never touches the SQLite file; only the seal blob does.
    """

    def __init__(
        self,
        path: str,
        passphrase: str,
        *,
        anchor_path: Optional[str] = None,
    ):
        if path == ":memory:":
            raise ValueError("SealedStore needs a filesystem path to seal")
        if len(passphrase) < MIN_PASSPHRASE:
            raise ValueError(
                f"seal passphrase must be at least {MIN_PASSPHRASE} characters"
            )
        self.seal_path = path
        self._passphrase = passphrase
        self._tmpdir = tempfile.TemporaryDirectory(prefix="sigil-seal-")
        self._plain_path = os.path.join(self._tmpdir.name, "store.db")
        if os.path.exists(path):
            with open(path, "rb") as handle:
                blob = handle.read()
            if not blob.startswith(MAGIC):
                raise ValueError(
                    f"{path} exists but is not a Sigil sealed store; "
                    "refusing to treat a cleartext DB as sealed"
                )
            plain = unseal_bytes(blob, passphrase)
            with open(self._plain_path, "wb") as handle:
                handle.write(plain)
        super().__init__(self._plain_path, anchor_path=anchor_path)

    def flush(self) -> str:
        """Checkpoint WAL, read plaintext bytes, write sealed blob to seal_path."""
        with self._write_lock:
            self.conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
            self.conn.commit()
            # Copy via SQLite backup so we don't race readers mid-page.
            dest_path = os.path.join(self._tmpdir.name, "export.db")
            if os.path.exists(dest_path):
                os.unlink(dest_path)
            dest = sqlite3.connect(dest_path)
            try:
                self.conn.backup(dest)
                dest.commit()
            finally:
                dest.close()
            with open(dest_path, "rb") as handle:
                plain = handle.read()
            blob = seal_bytes(plain, self._passphrase)
            directory = os.path.dirname(os.path.abspath(self.seal_path))
            if directory:
                os.makedirs(directory, exist_ok=True)
            fd, tmp = tempfile.mkstemp(
                prefix=".sigil-seal-", suffix=".tmp", dir=directory or "."
            )
            try:
                with os.fdopen(fd, "wb") as handle:
                    handle.write(blob)
                    handle.flush()
                    os.fsync(handle.fileno())
                os.replace(tmp, self.seal_path)
            except Exception:
                try:
                    os.unlink(tmp)
                except OSError:
                    pass
                raise
            os.chmod(self.seal_path, 0o600)
        return self.seal_path

    def close(self) -> None:
        """Flush the seal and release the temp working copy."""
        try:
            self.flush()
        finally:
            try:
                self.conn.close()
            except Exception:
                pass
            self._tmpdir.cleanup()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        self.close()
        return False