encryption

encryption

Passphrase plumbing and SQLCipher helpers for at-rest credential encryption.

Walks the five-tier resolution chain — session-unlock file → systemd-creds → OS keyring → passphrase_command helper → interactive prompt — and exposes the SQLCipher open / rekey / migrate primitives the rest of the package builds on. The tier vocabulary lives in tiers; resolve_passphrase documents the chain order; open_sqlcipher is the only entry point that ever calls sqlcipher3.connect.

The setup-time plaintext→SQLCipher migration (deprecated in 0.8.0, removed in 0.9.0) lives at the bottom of the file; nothing in the runtime chain touches it.

KEYRING_SERVICE = 'terok-sandbox' module-attribute

KEYRING_USERNAME = 'credentials-db' module-attribute

__all__ = ['KEYRING_SERVICE', 'KEYRING_USERNAME', 'NoPassphraseError', 'PassphraseTier', 'WrongPassphraseError', 'encrypt_in_place', 'forget_passphrase_in_keyring', 'generate_passphrase', 'is_plaintext_sqlite', 'keyring_backend_available', 'load_passphrase_from_command', 'load_passphrase_from_file', 'load_passphrase_from_keyring', 'open_sqlcipher', 'open_sqlcipher_via_chain', 'prompt_new_passphrase', 'prompt_passphrase', 'rekey_in_place', 'resolve_passphrase', 'resolve_passphrase_with_source', 'store_passphrase_in_keyring'] module-attribute

PassphraseTier

Bases: StrEnum

Where the vault passphrase can live, in resolution-chain priority order.

A StrEnum so members compare, hash, and serialise as their plain string values — status JSON, config knobs, and CLI arguments all speak the same vocabulary without conversion shims.

SESSION_FILE = 'session-file' class-attribute instance-attribute

Tmpfs session-unlock file — RAM-backed, cleared on reboot.

SYSTEMD_CREDS = 'systemd-creds' class-attribute instance-attribute

Sealed machine-bound credential (TPM2 / host key); needs systemd ≥ 257.

KEYRING = 'keyring' class-attribute instance-attribute

OS keyring entry, unlocked together with the login session.

PASSPHRASE_COMMAND = 'passphrase-command' class-attribute instance-attribute

Operator-supplied helper command that prints the passphrase (pass show …, bw get …, op read …, cloud secret CLIs).

PROMPT = 'prompt' class-attribute instance-attribute

Interactive TTY entry — stores nothing, ever.

durable property

True iff the tier survives a reboot.

A volatile tier resolving on top of a durable one is shadowing it — the vault silently reads the copy that dies on the next boot.

provisionable property

True iff a value can be written into the tier programmatically.

passphrase-command is the counter-example: the secret lives in a store the operator owns (pass / bitwarden / a cloud secret manager), so the sandbox can read it but never write it. prompt stores nothing at all.

chooser_offered property

True iff the interactive setup chooser lists the tier.

systemd-creds is deliberately not offered — it auto-selects whenever the host supports it, so listing it would only add a dead option to the menu.

NoPassphraseError

Bases: RuntimeError

No SQLCipher passphrase resolved — the DB cannot be opened.

WrongPassphraseError

Bases: RuntimeError

SQLCipher could not decrypt the DB — passphrase doesn't match its encryption key.

TierPresence(source, present, detail) dataclass

Whether one passphrase-chain tier currently holds material — for vault status.

A diagnostic, non-short-circuiting counterpart to resolve_passphrase_with_source: that walker stops at the first tier that resolves, so it can only ever name the winner. vault status needs the whole chain to show when a high-priority tier (typically the session file) is shadowing a durable tier underneath — the operator's "why is my TPM2 box reading a RAM-backed file?" question.

source instance-attribute

present instance-attribute

detail instance-attribute

open_sqlcipher_via_chain(db_path, *, passphrase_file=None, systemd_creds_file=None, use_keyring=False, passphrase_command=None, prompt_on_tty=False, **connect_kwargs)

Resolve the passphrase through the runtime chain and open db_path.

Raises NoPassphraseError when the chain yields nothing. prompt_on_tty turns on the interactive fallback for CLI consumers; daemons leave it False.

Source code in src/terok_sandbox/vault/store/encryption.py

def open_sqlcipher_via_chain(
    db_path: str | Path,
    *,
    passphrase_file: Path | None = None,
    systemd_creds_file: Path | None = None,
    use_keyring: bool = False,
    passphrase_command: str | None = None,
    prompt_on_tty: bool = False,
    **connect_kwargs: Any,
) -> Any:
    """Resolve the passphrase through the runtime chain and open *db_path*.

    Raises [`NoPassphraseError`][terok_sandbox.vault.store.encryption.NoPassphraseError]
    when the chain yields nothing.  *prompt_on_tty* turns on the
    interactive fallback for CLI consumers; daemons leave it ``False``.
    """
    passphrase = resolve_passphrase(
        passphrase_file=passphrase_file,
        systemd_creds_file=systemd_creds_file,
        use_keyring=use_keyring,
        passphrase_command=passphrase_command,
        prompt_on_tty=prompt_on_tty,
    )
    if passphrase is None:
        raise NoPassphraseError(f"no SQLCipher passphrase available for {db_path}")
    return open_sqlcipher(db_path, passphrase, **connect_kwargs)

resolve_passphrase_with_source(*, passphrase_file=None, systemd_creds_file=None, use_keyring=False, passphrase_command=None, prompt_on_tty=False)

Walk the runtime resolution chain; return (passphrase, source).

Single source of truth for the resolution order — see resolve_passphrase for the tier semantics. Both elements of the tuple are None when no tier had a passphrase.

The source half feeds a TUI/CLI status display — keep the labels stable, callers dispatch on them.

Source code in src/terok_sandbox/vault/store/encryption.py

def resolve_passphrase_with_source(
    *,
    passphrase_file: Path | None = None,
    systemd_creds_file: Path | None = None,
    use_keyring: bool = False,
    passphrase_command: str | None = None,
    prompt_on_tty: bool = False,
) -> tuple[str | None, PassphraseTier | None]:
    """Walk the runtime resolution chain; return ``(passphrase, source)``.

    Single source of truth for the resolution order — see
    [`resolve_passphrase`][terok_sandbox.vault.store.encryption.resolve_passphrase]
    for the tier semantics.  Both elements of the tuple are ``None``
    when no tier had a passphrase.

    The source half feeds a TUI/CLI status display — keep the labels
    stable, callers dispatch on them.
    """
    # Truthy checks throughout: an empty string anywhere in the chain
    # is SQLCipher's no-encryption sentinel; treat it as "not present"
    # rather than letting it overrule a real later tier.
    if passphrase_file is not None:
        file_pw = load_passphrase_from_file(passphrase_file)
        if file_pw:
            return file_pw, PassphraseTier.SESSION_FILE
    if systemd_creds_file is not None and systemd_creds_file.is_file():
        sealed_pw = _systemd_creds.unseal(systemd_creds_file)
        if sealed_pw:
            return sealed_pw, PassphraseTier.SYSTEMD_CREDS
        # Fail closed: silently falling through would demote a
        # machine-bound tier to keyring / plaintext-on-disk without
        # the operator's knowledge.
        raise WrongPassphraseError(
            f"sealed systemd-creds credential present at {systemd_creds_file}"
            " but could not be unsealed"
        )
    if use_keyring:
        keyring_pw = load_passphrase_from_keyring()
        if keyring_pw:
            return keyring_pw, PassphraseTier.KEYRING
    if passphrase_command:
        cmd_pw = load_passphrase_from_command(passphrase_command)
        if cmd_pw:
            return cmd_pw, PassphraseTier.PASSPHRASE_COMMAND
        # Fail closed for the same reason as systemd-creds above; the
        # command string itself is omitted because operators sometimes
        # inline AWS ARNs / vault paths there and this exception reaches
        # doctor output and journals.
        raise WrongPassphraseError(
            "passphrase_command produced no passphrase; run it manually to diagnose"
            " (see WARNING in the vault journal)"
        )
    if prompt_on_tty and sys.stdin.isatty():
        return prompt_passphrase(), PassphraseTier.PROMPT
    return None, None

resolve_passphrase(*, passphrase_file=None, systemd_creds_file=None, use_keyring=False, passphrase_command=None, prompt_on_tty=False)

Walk the runtime resolution chain; return None if nothing has it.

Order:

1. passphrase_file — session-unlock tmpfs file (cleared on reboot).
2. systemd_creds_file — sealed credential decrypted via systemd-creds(1). Machine-bound (TPM2 or host key), survives reboot, no OS keyring required. See terok_sandbox.vault.store.systemd_creds.
3. OS keyring — only when use_keyring is true; off by default because Linux Secret Service grants access per-collection, not per-item.
4. passphrase_command — operator-supplied shell command (pass show …, bw get, op read, cloud secret-manager CLIs). Delegates retrieval without per-backend integration code, same shape as git config credential.helper or BORG_PASSCOMMAND. Configured-but-broken fails closed so a misbehaving helper can't silently demote security to a weaker tier.
5. Interactive prompt — only when prompt_on_tty and sys.stdin.isatty().

passphrase_command is threaded through as a parameter rather than read here so this module stays free of any dependency on the sandbox config layer — the config module already imports from credentials.db, and the back-edge would close a tach cycle.

Source code in src/terok_sandbox/vault/store/encryption.py

def resolve_passphrase(
    *,
    passphrase_file: Path | None = None,
    systemd_creds_file: Path | None = None,
    use_keyring: bool = False,
    passphrase_command: str | None = None,
    prompt_on_tty: bool = False,
) -> str | None:
    """Walk the runtime resolution chain; return ``None`` if nothing has it.

    Order:

    1. *passphrase_file* — session-unlock tmpfs file (cleared on reboot).
    2. *systemd_creds_file* — sealed credential decrypted via
       ``systemd-creds(1)``.  Machine-bound (TPM2 or host key), survives
       reboot, no OS keyring required.  See
       [`terok_sandbox.vault.store.systemd_creds`][terok_sandbox.vault.store.systemd_creds].
    3. OS keyring — only when *use_keyring* is true; off by default because
       Linux Secret Service grants access per-collection, not per-item.
    4. *passphrase_command* — operator-supplied shell command
       (``pass show …``, ``bw get``, ``op read``, cloud secret-manager
       CLIs).  Delegates retrieval without per-backend integration code,
       same shape as ``git config credential.helper`` or
       ``BORG_PASSCOMMAND``.  Configured-but-broken fails closed so a
       misbehaving helper can't silently demote security to a weaker tier.
    5. Interactive prompt — only when *prompt_on_tty* and ``sys.stdin.isatty()``.

    *passphrase_command* is threaded through as a parameter rather than
    read here so this module stays free of any dependency on the
    sandbox config layer — the config module already imports from
    credentials.db, and the back-edge would close a tach cycle.
    """
    passphrase, _source = resolve_passphrase_with_source(
        passphrase_file=passphrase_file,
        systemd_creds_file=systemd_creds_file,
        use_keyring=use_keyring,
        passphrase_command=passphrase_command,
        prompt_on_tty=prompt_on_tty,
    )
    return passphrase

probe_passphrase_chain(*, passphrase_file=None, systemd_creds_file=None, use_keyring=False, passphrase_command=None)

Report per-tier presence across the resolution chain without short-circuiting.

Presence is judged from material on hand, not by resolving the secret: the sealed systemd-creds credential is never unsealed and the passphrase_command is never executed (both can be slow or have side effects), so their mere configuration counts as present. The session-file and keyring tiers are cheap to read, so those are probed for real. Tiers appear in resolution order; the first present one is the tier that would unlock the vault. The interactive prompt tier is omitted — it stores nothing, so it can neither be "present" nor shadow anything.

Source code in src/terok_sandbox/vault/store/encryption.py

def probe_passphrase_chain(
    *,
    passphrase_file: Path | None = None,
    systemd_creds_file: Path | None = None,
    use_keyring: bool = False,
    passphrase_command: str | None = None,
) -> tuple[TierPresence, ...]:
    """Report per-tier presence across the resolution chain without short-circuiting.

    Presence is judged from *material on hand*, not by resolving the
    secret: the sealed systemd-creds credential is never unsealed and
    the ``passphrase_command`` is never executed (both can be slow or
    have side effects), so their mere configuration counts as present.
    The session-file and keyring tiers are cheap to read, so those are
    probed for real.  Tiers appear in resolution order; the first
    ``present`` one is the tier that would unlock the vault.  The
    interactive ``prompt`` tier is omitted — it stores nothing, so it
    can neither be "present" nor shadow anything.
    """
    session_value = load_passphrase_from_file(passphrase_file) if passphrase_file else None
    session_detail = str(passphrase_file) if passphrase_file else "no session file"
    # A file that exists but yields nothing is a fault (permissions,
    # SELinux, empty write), not a locked vault — say so in the detail
    # line; the silent variant cost us a debugging session already.
    if passphrase_file is not None and session_value is None and passphrase_file.exists():
        session_detail = f"{passphrase_file} (exists but unreadable or empty)"
    return (
        TierPresence(
            PassphraseTier.SESSION_FILE,
            bool(session_value),
            session_detail,
        ),
        TierPresence(
            PassphraseTier.SYSTEMD_CREDS,
            bool(systemd_creds_file and systemd_creds_file.is_file()),
            _systemd_creds_detail(systemd_creds_file),
        ),
        TierPresence(
            PassphraseTier.KEYRING,
            # Truthy, not ``is not None``: an empty string is the resolver's
            # "no passphrase" sentinel, so status must treat it as absent too.
            use_keyring and bool(load_passphrase_from_keyring()),
            "OS keyring" if use_keyring else "use_keyring off",
        ),
        TierPresence(
            PassphraseTier.PASSPHRASE_COMMAND,
            bool(passphrase_command),
            "configured (not executed)" if passphrase_command else "not configured",
        ),
    )

load_passphrase_from_file(path)

Return the passphrase stored at path, or None if absent or unreadable.

An absent file is the normal locked state and stays silent. Any other OSError (EACCES from a permissions slip, an SELinux denial, an unmounted tmpfs) also degrades to None so the chain can fall through — but it logs a warning first: without the log a blocked read is indistinguishable from "locked" on every surface, which buries the actual fault.

Source code in src/terok_sandbox/vault/store/encryption.py

def load_passphrase_from_file(path: Path) -> str | None:
    """Return the passphrase stored at *path*, or ``None`` if absent or unreadable.

    An absent file is the normal locked state and stays silent.  Any
    *other* ``OSError`` (EACCES from a permissions slip, an SELinux
    denial, an unmounted tmpfs) also degrades to ``None`` so the chain
    can fall through — but it logs a warning first: without the log a
    blocked read is indistinguishable from "locked" on every surface,
    which buries the actual fault.
    """
    try:
        return path.read_text(encoding="utf-8").rstrip("\n") or None
    except FileNotFoundError:
        return None
    except OSError as exc:
        _logger.warning("session passphrase file %s exists but is unreadable: %s", path, exc)
        return None

load_passphrase_from_keyring()

Return the keyring-stored passphrase, or None if no backend is reachable.

Source code in src/terok_sandbox/vault/store/encryption.py

def load_passphrase_from_keyring() -> str | None:
    """Return the keyring-stored passphrase, or ``None`` if no backend is reachable."""
    try:
        import keyring  # noqa: PLC0415

        return keyring.get_password(KEYRING_SERVICE, KEYRING_USERNAME)
    except Exception:  # noqa: BLE001
        return None

keyring_backend_available()

Return True iff a usable OS keyring backend is reachable.

Availability probe for setup frontends (the TUI tier chooser) deciding whether to offer the keyring tier at all. A probe, not a guarantee — the definitive answer stays with store_passphrase_in_keyring's return value at provisioning time. The fail and null backends both answer "no": they accept calls but hold nothing.

Source code in src/terok_sandbox/vault/store/encryption.py

def keyring_backend_available() -> bool:
    """Return ``True`` iff a usable OS keyring backend is reachable.

    Availability probe for setup frontends (the TUI tier chooser)
    deciding whether to *offer* the keyring tier at all.  A probe, not
    a guarantee — the definitive answer stays with
    [`store_passphrase_in_keyring`][terok_sandbox.vault.store.encryption.store_passphrase_in_keyring]'s
    return value at provisioning time.  The ``fail`` and ``null``
    backends both answer "no": they accept calls but hold nothing.
    """
    try:
        import keyring  # noqa: PLC0415
        from keyring.backends import fail, null  # noqa: PLC0415

        return not isinstance(keyring.get_keyring(), (fail.Keyring, null.Keyring))
    except Exception:  # noqa: BLE001
        return False

store_passphrase_in_keyring(passphrase)

Persist passphrase in the OS keyring; return True on success.

Refuses to store an empty value — SQLCipher interprets it as "no encryption", and a later resolve hit on a blank keyring entry would silently open the DB plaintext.

Source code in src/terok_sandbox/vault/store/encryption.py

def store_passphrase_in_keyring(passphrase: str) -> bool:
    """Persist *passphrase* in the OS keyring; return ``True`` on success.

    Refuses to store an empty value — SQLCipher interprets it as
    "no encryption", and a later resolve hit on a blank keyring entry
    would silently open the DB plaintext.
    """
    if not passphrase:
        raise ValueError("refusing to store an empty passphrase in the keyring")
    try:
        import keyring  # noqa: PLC0415

        keyring.set_password(KEYRING_SERVICE, KEYRING_USERNAME, passphrase)
        return True
    except Exception:  # noqa: BLE001
        return False

forget_passphrase_in_keyring()

Remove the keyring entry; return True on success.

Source code in src/terok_sandbox/vault/store/encryption.py

def forget_passphrase_in_keyring() -> bool:
    """Remove the keyring entry; return ``True`` on success."""
    try:
        import keyring  # noqa: PLC0415

        keyring.delete_password(KEYRING_SERVICE, KEYRING_USERNAME)
        return True
    except Exception:  # noqa: BLE001
        return False

load_passphrase_from_command(command, *, timeout=_PASSPHRASE_COMMAND_TIMEOUT_S)

Run command, return its stdout with the trailing newline removed, or None on any failure.

Same shape as the other tier primitives (load_passphrase_from_file, load_passphrase_from_keyring): silent on every failure path so the resolver can decide whether None means "skip this tier" or "fail closed". Diagnostic detail (parse error, exec failure, non-zero exit, helper stderr, timeout) is logged at WARNING so operators can triage their helper from the invoking command's output (or the per-container supervisor log under <state_root>/logs/) without us crashing the chain walk.

Same vocabulary as git config credential.helper, ssh pinentry, BORG_PASSCOMMAND: one field plugs any credential backend into the resolver — pass show …, bw get password …, op read op://…, vault kv get -field=passphrase …, aws secretsmanager get-secret-value … — without per-backend integration code in the sandbox.

Source code in src/terok_sandbox/vault/store/encryption.py

def load_passphrase_from_command(
    command: str, *, timeout: float = _PASSPHRASE_COMMAND_TIMEOUT_S
) -> str | None:
    """Run *command*, return its stdout with the trailing newline removed, or ``None`` on any failure.

    Same shape as the other tier primitives ([`load_passphrase_from_file`][terok_sandbox.vault.store.encryption.load_passphrase_from_file],
    [`load_passphrase_from_keyring`][terok_sandbox.vault.store.encryption.load_passphrase_from_keyring]):
    silent on every failure path so the resolver can decide whether
    ``None`` means "skip this tier" or "fail closed".  Diagnostic
    detail (parse error, exec failure, non-zero exit, helper stderr,
    timeout) is logged at WARNING so operators can triage their helper
    from the invoking command's output (or the per-container supervisor
    log under ``<state_root>/logs/``) without us crashing the chain
    walk.

    Same vocabulary as ``git config credential.helper``, ssh pinentry,
    ``BORG_PASSCOMMAND``: one field plugs any credential backend into
    the resolver — ``pass show …``, ``bw get password …``,
    ``op read op://…``, ``vault kv get -field=passphrase …``,
    ``aws secretsmanager get-secret-value …`` — without per-backend
    integration code in the sandbox.
    """
    try:
        argv = shlex.split(command)
    except ValueError as exc:
        _logger.warning("passphrase_command shlex parse failed: %s", exc)
        return None
    if not argv:
        return None
    try:
        result = subprocess.run(  # noqa: S603 — argv is operator-configured  # nosec B603 — argv is a fixed list controlled by this module — argv is a fixed list controlled by this module
            argv, capture_output=True, text=True, timeout=timeout, check=False
        )
    except OSError as exc:
        _logger.warning("passphrase_command %r failed to spawn: %s", argv[0], exc)
        return None
    except subprocess.TimeoutExpired:
        _logger.warning("passphrase_command %r timed out after %.0fs", argv[0], timeout)
        return None
    if result.returncode != 0:
        _logger.warning(
            "passphrase_command %r exited %d: %s",
            argv[0],
            result.returncode,
            result.stderr.strip() or "(no stderr)",
        )
        return None
    # rstrip only the line ending the helper appends — leading/trailing
    # whitespace inside the passphrase is legitimate secret material and
    # must reach SQLCipher verbatim.
    passphrase = result.stdout.rstrip("\r\n")
    return passphrase or None

prompt_passphrase(*, confirm=False)

Read a passphrase from the controlling TTY with *-masked echo.

Mirrors the _prompt_api_key helper in terok_executor.credentials.auth: prompt_toolkit.prompt(is_password=True) for the TTY path — proper terminal raw-mode handling, Ctrl+C raises KeyboardInterrupt cleanly, every character is masked. Non-TTY input (e.g. terok-sandbox credentials encrypt-db < passphrase.txt) falls back to a plain readline so pipe-fed automation still works.

Empty entries are SQLCipher's no-encryption sentinel and never return a blank string. In confirm mode (setup-time provisioning of a brand-new passphrase) hitting Enter is treated as "generate one for me": a fresh random passphrase is minted, echoed once so the operator can copy it out, and returned. In single-shot mode (unlocking an existing DB) an empty entry raises — generating here would produce a wrong key that fails to decrypt the DB.

Source code in src/terok_sandbox/vault/store/encryption.py

def prompt_passphrase(*, confirm: bool = False) -> str:
    """Read a passphrase from the controlling TTY with ``*``-masked echo.

    Mirrors the ``_prompt_api_key`` helper in [`terok_executor.credentials.auth`][terok_executor.credentials.auth]:
    ``prompt_toolkit.prompt(is_password=True)`` for the TTY path —
    proper terminal raw-mode handling, ``Ctrl+C`` raises
    ``KeyboardInterrupt`` cleanly, every character is masked.  Non-TTY
    input (e.g. ``terok-sandbox credentials encrypt-db < passphrase.txt``)
    falls back to a plain ``readline`` so pipe-fed automation still
    works.

    Empty entries are SQLCipher's no-encryption sentinel and never
    return a blank string.  In *confirm* mode (setup-time provisioning
    of a brand-new passphrase) hitting ``Enter`` is treated as
    "generate one for me": a fresh random passphrase is minted, echoed
    once so the operator can copy it out, and returned.  In single-shot
    mode (unlocking an existing DB) an empty entry raises — generating
    here would produce a wrong key that fails to decrypt the DB.
    """
    if sys.stdin.isatty():
        if confirm:
            typed = prompt_new_passphrase()
            if typed is not None:
                return typed
            # Empty + confirm = "mint one for me".  Write to
            # ``/dev/tty`` (not stdout) so a redirected install
            # — ``terok-sandbox setup > install.log`` or CI —
            # can't capture the recovery key.  ``commands._announce_generated_passphrase``
            # does the same thing for non-``prompt_passphrase``
            # paths; this is the foundation-layer mirror (we
            # can't import from the surface layer per tach).
            passphrase = generate_passphrase()
            _write_to_controlling_tty(
                f"\nVault passphrase: {passphrase}\n"
                "  Write this down — it's your recovery key for rebuilds and other hosts.\n"
            )
            return passphrase
        from prompt_toolkit import prompt as ptk_prompt  # noqa: PLC0415

        try:
            passphrase = ptk_prompt("credentials.db passphrase: ", is_password=True).strip()
        except (KeyboardInterrupt, EOFError):
            raise SystemExit("passphrase entry cancelled.") from None
    else:
        passphrase = sys.stdin.readline().rstrip("\n")
    if not passphrase:
        raise ValueError("empty passphrase")
    return passphrase

prompt_new_passphrase()

Read a typed-and-confirmed new passphrase from the TTY, or None on empty entry.

The entry half of prompt_passphrase's confirm mode, split out so flows that mint-and-announce on their own schedule (vault passphrase change reveals only after the rekey succeeded) can reuse the exact same typed-entry experience: masked echo, a confirmation read, a mismatch error. None means the operator hit Enter on the first prompt — the established "generate one for me" gesture.

Source code in src/terok_sandbox/vault/store/encryption.py

def prompt_new_passphrase() -> str | None:
    """Read a typed-and-confirmed *new* passphrase from the TTY, or ``None`` on empty entry.

    The entry half of [`prompt_passphrase`][terok_sandbox.vault.store.encryption.prompt_passphrase]'s
    ``confirm`` mode, split out so flows that mint-and-announce on
    their own schedule (``vault passphrase change`` reveals only after
    the rekey succeeded) can reuse the exact same typed-entry
    experience: masked echo, a confirmation read, a mismatch error.
    ``None`` means the operator hit ``Enter`` on the first prompt —
    the established "generate one for me" gesture.
    """
    from prompt_toolkit import prompt as ptk_prompt  # noqa: PLC0415

    try:
        passphrase = ptk_prompt("credentials.db passphrase: ", is_password=True).strip()
        if not passphrase:
            return None
        again = ptk_prompt("confirm passphrase:        ", is_password=True).strip()
        if passphrase != again:
            raise ValueError("passphrases do not match")
        return passphrase
    except (KeyboardInterrupt, EOFError):
        raise SystemExit("passphrase entry cancelled.") from None

open_sqlcipher(db_path, passphrase, **connect_kwargs)

Return a sqlcipher3 connection with passphrase applied.

Rejects an empty passphrase at the lowest level — set_key("") is SQLCipher's "open me plaintext" sentinel and would silently produce or read an unencrypted DB. All higher-level call paths already screen for empties; this is the load-bearing guard.

Source code in src/terok_sandbox/vault/store/encryption.py

def open_sqlcipher(db_path: str | Path, passphrase: str, **connect_kwargs: Any) -> Any:
    """Return a sqlcipher3 connection with *passphrase* applied.

    Rejects an empty passphrase at the lowest level — ``set_key("")``
    is SQLCipher's "open me plaintext" sentinel and would silently
    produce or read an unencrypted DB.  All higher-level call paths
    already screen for empties; this is the load-bearing guard.
    """
    if not passphrase:
        raise ValueError("empty passphrase would disable SQLCipher encryption")
    import sqlcipher3  # noqa: PLC0415

    conn = sqlcipher3.connect(str(db_path), **connect_kwargs)
    conn.set_key(passphrase)
    conn.execute("PRAGMA cipher_compatibility = 4")
    return conn

generate_passphrase()

Return a freshly-randomised url-safe passphrase.

Source code in src/terok_sandbox/vault/store/encryption.py

def generate_passphrase() -> str:
    """Return a freshly-randomised url-safe passphrase."""
    return secrets.token_urlsafe(_GENERATED_PASSPHRASE_BYTES)

rekey_in_place(db_path, old_passphrase, new_passphrase)

Re-encrypt db_path under new_passphrase via SQLCipher PRAGMA rekey.

The change-passphrase counterpart of encrypt_in_place — same journal discipline, different starting point: an already-encrypted DB instead of a legacy plaintext one, re-encrypted page by page inside SQLCipher's own journaled transaction (no temp copy, no backup — the operator's off-host passphrase copy is the recovery story).

The WAL is drained and journaling switched to DELETE first: WAL frames are encrypted with the old key, so rekeying around a populated WAL would leave frames the new key cannot read. Both steps report contention through their result rows rather than by raising — wal_checkpoint returns a busy flag and journal_mode echoes the old mode when it couldn't switch — so each is verified explicitly, and a live per-container supervisor still holding the DB surfaces as database is locked before anything is modified.

Raises WrongPassphraseError when old_passphrase doesn't open the DB and ValueError on an empty new passphrase (SQLCipher's no-encryption sentinel).

Source code in src/terok_sandbox/vault/store/encryption.py

def rekey_in_place(db_path: Path, old_passphrase: str, new_passphrase: str) -> None:
    """Re-encrypt *db_path* under *new_passphrase* via SQLCipher ``PRAGMA rekey``.

    The change-passphrase counterpart of
    [`encrypt_in_place`][terok_sandbox.vault.store.encryption.encrypt_in_place]
    — same journal discipline, different starting point: an
    already-encrypted DB instead of a legacy plaintext one, re-encrypted
    page by page inside SQLCipher's own journaled transaction (no temp
    copy, no backup — the operator's off-host passphrase copy is the
    recovery story).

    The WAL is drained and journaling switched to ``DELETE`` first: WAL
    frames are encrypted with the *old* key, so rekeying around a
    populated WAL would leave frames the new key cannot read.  Both
    steps report contention through their *result rows* rather than by
    raising — ``wal_checkpoint`` returns a busy flag and
    ``journal_mode`` echoes the old mode when it couldn't switch — so
    each is verified explicitly, and a live per-container supervisor
    still holding the DB surfaces as ``database is locked`` *before*
    anything is modified.

    Raises [`WrongPassphraseError`][terok_sandbox.vault.store.encryption.WrongPassphraseError]
    when *old_passphrase* doesn't open the DB and [`ValueError`][ValueError]
    on an empty new passphrase (SQLCipher's no-encryption sentinel).
    """
    if not new_passphrase:
        raise ValueError("empty passphrase would disable SQLCipher encryption")
    import sqlcipher3  # noqa: PLC0415

    conn = open_sqlcipher(db_path, old_passphrase)
    try:
        try:
            # A wrong key only surfaces on the first page read — force
            # one before touching anything.
            conn.execute("SELECT count(*) FROM sqlite_master")
        except sqlcipher3.DatabaseError as exc:
            if "file is not a database" not in str(exc):
                raise  # a real fault ("database is locked", I/O) — not a key mismatch
            raise WrongPassphraseError(f"the current passphrase does not open {db_path}") from exc
        busy, _log_pages, _moved_pages = conn.execute("PRAGMA wal_checkpoint(FULL)").fetchone()
        if busy:
            raise RuntimeError(
                f"database is locked — cannot drain the WAL of {db_path};"
                " another connection is still holding it"
            )
        (mode,) = conn.execute("PRAGMA journal_mode=DELETE").fetchone()
        if str(mode).lower() != "delete":
            raise RuntimeError(
                f"database is locked — cannot take exclusive hold of {db_path};"
                " another connection is still open"
            )
        # PRAGMA statements cannot take bound parameters; single-quote
        # doubling is SQL's exact string-literal escape, valid for any
        # passphrase content.
        conn.execute("PRAGMA rekey = '{}'".format(new_passphrase.replace("'", "''")))
        conn.execute("PRAGMA journal_mode=WAL")
        # No sidecar cleanup after this point: the DELETE-mode switch
        # already removed the old-key WAL, and an unlink after close
        # would race a fresh supervisor's brand-new (new-key) sidecars.
    finally:
        conn.close()

is_plaintext_sqlite(db_path)

Return True if db_path is a legacy plaintext sqlite DB.

Stdlib sqlite refuses to open SQLCipher files with DatabaseError: file is not a database; a successful PRAGMA quick_check means the file is plain sqlite. Used only by the one-shot setup migration — not on any runtime open path.

Source code in src/terok_sandbox/vault/store/encryption.py

def is_plaintext_sqlite(db_path: Path) -> bool:
    """Return ``True`` if *db_path* is a legacy plaintext sqlite DB.

    Stdlib sqlite refuses to open SQLCipher files with ``DatabaseError:
    file is not a database``; a successful ``PRAGMA quick_check`` means
    the file is plain sqlite.  Used only by the one-shot setup
    migration — not on any runtime open path.
    """
    if not db_path.exists() or db_path.stat().st_size == 0:
        return False
    try:
        conn = sqlite3.connect(str(db_path))
        try:
            conn.execute("PRAGMA quick_check").fetchone()
        finally:
            conn.close()
    except sqlite3.DatabaseError:
        return False
    return True

encrypt_in_place(db_path, passphrase)

Convert plaintext db_path into a SQLCipher-encrypted DB.

Deprecated in 0.8.0; scheduled for removal in 0.9.0. After removal, this function and its CLI surface (terok-sandbox credentials encrypt-db) disappear — installs older than 0.8.0 must migrate before upgrading past 0.9.0.

Atomic: a crash between export and rename leaves the original plaintext file untouched, so a re-run starts cleanly.

WAL-aware: the legacy DB may have been opened in WAL mode (the daemon sets journal_mode=WAL on every connection), so its pages can live in .db-wal rather than the main file. Before exporting we force a full checkpoint and switch to DELETE journaling, then unlink the -wal / -shm / -journal sidecars; otherwise plaintext secrets would survive the migration in the leftover sidecars even after the main file is encrypted.

Permission-tight: the temp file is created up-front at 0o600 so SQLCipher's ATTACH doesn't materialise a world-readable encrypted DB under a permissive umask.

Source code in src/terok_sandbox/vault/store/encryption.py

def encrypt_in_place(db_path: Path, passphrase: str) -> None:
    """Convert plaintext *db_path* into a SQLCipher-encrypted DB.

    Deprecated in 0.8.0; scheduled for removal in 0.9.0.  After
    removal, this function and its CLI surface
    (``terok-sandbox credentials encrypt-db``) disappear — installs
    older than 0.8.0 must migrate before upgrading past 0.9.0.

    Atomic: a crash between export and rename leaves the original
    plaintext file untouched, so a re-run starts cleanly.

    WAL-aware: the legacy DB may have been opened in WAL mode (the
    daemon sets ``journal_mode=WAL`` on every connection), so its
    pages can live in ``.db-wal`` rather than the main file.  Before
    exporting we force a full checkpoint and switch to ``DELETE``
    journaling, then unlink the ``-wal`` / ``-shm`` / ``-journal``
    sidecars; otherwise plaintext secrets would survive the migration
    in the leftover sidecars even after the main file is encrypted.

    Permission-tight: the temp file is created up-front at 0o600 so
    SQLCipher's ``ATTACH`` doesn't materialise a world-readable
    encrypted DB under a permissive umask.
    """
    if not passphrase:
        raise ValueError("empty passphrase would produce a plaintext DB")
    if not db_path.exists():
        raise FileNotFoundError(db_path)

    tmp_path = db_path.with_suffix(db_path.suffix + ".encrypting")
    tmp_path.unlink(missing_ok=True)
    # Materialise tmp_path at 0o600 before ATTACH so SQLCipher inherits
    # those bits instead of the umask default — the file is empty so
    # SQLCipher will populate it freely.
    os.close(os.open(tmp_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600))

    import sqlcipher3  # noqa: PLC0415

    try:
        conn = sqlcipher3.connect(str(db_path))
        try:
            # Drain WAL into the main file and stop journaling so the
            # subsequent sidecar unlink genuinely removes plaintext data.
            conn.execute("PRAGMA wal_checkpoint(FULL)")
            conn.execute("PRAGMA journal_mode=DELETE")

            conn.execute(
                "ATTACH DATABASE ? AS encrypted KEY ?",
                (str(tmp_path), passphrase),
            )
            conn.execute("PRAGMA encrypted.cipher_compatibility = 4")
            (result,) = conn.execute("SELECT sqlcipher_export('encrypted')").fetchone() or (None,)
            conn.execute("DETACH DATABASE encrypted")
        finally:
            conn.close()

        if result is not None and result != 0:
            raise RuntimeError(f"sqlcipher_export returned {result!r}")
    except BaseException:
        # Any failure between pre-create and replace must scrub the
        # ``.encrypting`` temp file and its sidecars so a re-run starts
        # clean.  ``BaseException`` covers SystemExit / KeyboardInterrupt
        # too — leaking a zero-byte tmp is the failure mode the user
        # actually hits ("database is locked" with a stale temp left
        # behind on disk).
        tmp_path.unlink(missing_ok=True)
        _unlink_sidecars(tmp_path)
        raise

    tmp_path.replace(db_path)
    # Sidecars under both names: plaintext leftovers from the legacy
    # connection (now next to the encrypted file) and any encrypted-side
    # sidecars that briefly accompanied the temp file.
    _unlink_sidecars(db_path)
    _unlink_sidecars(tmp_path)