encryption

encryption

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

Walks the five-tier resolution chain — systemd-creds → OS keyring → kernel 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_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.

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.

KERNEL_KEYRING = 'kernel-keyring' class-attribute instance-attribute

Linux kernel keyring (@u) — a volatile unlock cache in unswappable kernel memory, uid-scoped, cleared on logout. Sits below the zero-friction durable tiers but above the passphrase-command helper: it never shadows a tier that already unlocks non-interactively, yet still spares the operator from re-running a helper or re-typing a prompt on every open.

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 every tier that currently holds material, not just the one that would unlock the vault.

source instance-attribute

present instance-attribute

detail instance-attribute

open_sqlcipher_via_chain(db_path, *, 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,
    *,
    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(
        credentials_db=db_path,
        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(*, credentials_db, 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.

credentials_db is the vault the passphrase is for: it scopes the kernel-keyring lookup to that DB's key (see kernel_keyring.key_description), so the cache for one vault never resolves another's — pass the same path the caller is about to open.

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(
    *,
    credentials_db: str | Path,
    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.

    *credentials_db* is the vault the passphrase is *for*: it scopes the
    kernel-keyring lookup to that DB's key (see
    [`kernel_keyring.key_description`][terok_sandbox.vault.store.kernel_keyring.key_description]),
    so the cache for one vault never resolves another's — pass the same
    path the caller is about to open.

    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 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(allow_prompt=prompt_on_tty)
        if keyring_pw:
            return keyring_pw, PassphraseTier.KEYRING
    # Volatile unlock cache, below the zero-friction durable tiers and
    # above the helper: fail-*open* like the OS keyring above it — an
    # absent or expired key falls through rather than masking the
    # durable ``passphrase_command`` beneath.  Read via the module
    # namespace so tests can monkeypatch the session cache away.
    kernel_pw = _session_cache.load(credentials_db)
    if kernel_pw:
        return kernel_pw, PassphraseTier.KERNEL_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(*, credentials_db, 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. 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.
2. OS keyring — only when use_keyring is true; off by default because Linux Secret Service grants access per-collection, not per-item.
3. Session cache — the volatile unlock cache (terok_sandbox.vault.store.session_cache): the kernel keyring, or a tmpfs session file where the kernel facility is unusable. Consulted unconditionally (an absent/expired/unavailable cache just yields None); positioned so it never shadows a zero-friction durable tier above but still spares re-running the helper below.
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(
    *,
    credentials_db: str | Path,
    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. *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].
    2. OS keyring — only when *use_keyring* is true; off by default because
       Linux Secret Service grants access per-collection, not per-item.
    3. Session cache — the volatile unlock cache
       ([`terok_sandbox.vault.store.session_cache`][terok_sandbox.vault.store.session_cache]):
       the kernel keyring, or a tmpfs session file where the kernel
       facility is unusable.  Consulted unconditionally (an
       absent/expired/unavailable cache just yields ``None``);
       positioned so it never shadows a zero-friction durable tier
       above but still spares re-running the helper below.
    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(
        credentials_db=credentials_db,
        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(*, credentials_db, 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 keyring and kernel-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 never be "present".

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

def probe_passphrase_chain(
    *,
    credentials_db: str | Path,
    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 keyring and kernel-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 never be "present".
    """
    # Presence only — the status chain reports *that* a tier holds
    # material, never its value, so this must not read the passphrase.
    kernel_cached = _session_cache.is_cached(credentials_db)
    return (
        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.KERNEL_KEYRING,
            kernel_cached,
            _session_cache.backing_detail(cached=kernel_cached),
        ),
        TierPresence(
            PassphraseTier.PASSPHRASE_COMMAND,
            bool(passphrase_command),
            "configured (not executed)" if passphrase_command else "not configured",
        ),
    )

load_passphrase_from_keyring(*, allow_prompt=False)

Return the keyring-stored passphrase, or None when a read cannot succeed.

A read from a locked Secret Service collection triggers the desktop's unlock dialog and waits for the answer. That wait is legitimate exactly once: an interactive caller (allow_prompt) on a host with a graphical session, where the operator sees the dialog and answers or cancels it. Every other read — a status probe, a poll, a daemon, a headless or SSH session — skips a locked collection instead of waiting on a dialog nobody sees. A timeout guards the promptless read, so a misbehaving backend degrades this tier instead of freezing its caller.

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

def load_passphrase_from_keyring(*, allow_prompt: bool = False) -> str | None:
    """Return the keyring-stored passphrase, or ``None`` when a read cannot succeed.

    A read from a locked Secret Service collection triggers the
    desktop's unlock dialog and waits for the answer.  That wait is
    legitimate exactly once: an interactive caller (*allow_prompt*) on
    a host with a graphical session, where the operator sees the dialog
    and answers or cancels it.  Every other read — a status probe, a
    poll, a daemon, a headless or SSH session — skips a locked
    collection instead of waiting on a dialog nobody sees.  A timeout
    guards the promptless read, so a misbehaving backend degrades this
    tier instead of freezing its caller.
    """

    def _read() -> str | None:
        import keyring  # noqa: PLC0415

        return keyring.get_password(KEYRING_SERVICE, KEYRING_USERNAME)

    # The probe is always bounded — ``dbus_init`` can block exactly like
    # the read, and a wedged D-Bus must not block any caller.  Only the
    # interactive read is unbounded: the unlock dialog may legitimately
    # wait on the operator, and a timeout would cancel a dialog they
    # are looking at.
    try:
        blocked = _call_with_timeout(
            lambda: os_keyring_read_blocked(allow_prompt=allow_prompt),
            _KEYRING_READ_TIMEOUT_S,
        )
        if blocked is not None:
            return None
        return _read() if allow_prompt else _call_with_timeout(_read, _KEYRING_READ_TIMEOUT_S)
    except Exception:  # noqa: BLE001 — timeout or backend error: the tier degrades
        return None

os_keyring_read_blocked(*, allow_prompt=False)

Explain why an OS-keyring read would block or fail, or None when safe.

Only the Secret Service backend can block: a read from a locked collection triggers a D-Bus unlock prompt, and the prompt waits for a desktop dialog. This probe reads the lock state without a prompt. A locked collection does not block an allow_prompt read when a graphical session is present — the operator answers the dialog. A probe error (no D-Bus session, no Secret Service daemon) also makes the tier unusable and returns a reason. Non-D-Bus backends never prompt, so they pass.

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

def os_keyring_read_blocked(*, allow_prompt: bool = False) -> str | None:
    """Explain why an OS-keyring read would block or fail, or ``None`` when safe.

    Only the Secret Service backend can block: a read from a locked
    collection triggers a D-Bus unlock prompt, and the prompt waits for
    a desktop dialog.  This probe reads the lock state without a
    prompt.  A locked collection does not block an *allow_prompt* read
    when a graphical session is present — the operator answers the
    dialog.  A probe error (no D-Bus session, no Secret Service daemon)
    also makes the tier unusable and returns a reason.  Non-D-Bus
    backends never prompt, so they pass.
    """
    try:
        import keyring  # noqa: PLC0415
        from keyring.backends.SecretService import Keyring as _SecretService  # noqa: PLC0415

        if not isinstance(keyring.get_keyring(), _SecretService):
            return None
        import secretstorage  # noqa: PLC0415

        connection = secretstorage.dbus_init()
        try:
            collection = secretstorage.get_default_collection(connection)
            if collection.is_locked():
                if allow_prompt and _graphical_session_present():
                    return None
                return "OS keyring locked (unlock it in a desktop session, or use another tier)"
        finally:
            connection.close()
    except Exception as exc:  # noqa: BLE001
        return f"OS keyring unreachable ({type(exc).__name__})"
    return None

retire_keyring_worker()

Join the keyring worker so the process is single-threaded again.

A process that confines itself after its passphrase chain needs this: Landlock below ABI 8 restricts one thread, and the helper refuses a process with two. A worker abandoned in a wedged read cannot be joined and is left where it is — the confinement then reports the thread, which is the honest outcome. The next read starts a fresh worker.

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

def retire_keyring_worker() -> None:
    """Join the keyring worker so the process is single-threaded again.

    A process that confines itself after its passphrase chain needs
    this: Landlock below ABI 8 restricts one thread, and the helper
    refuses a process with two.  A worker abandoned in a wedged read
    cannot be joined and is left where it is — the confinement then
    reports the thread, which is the honest outcome.  The next read
    starts a fresh worker.
    """
    global _keyring_executor
    executor, _keyring_executor = _keyring_executor, None
    if executor is not None and not _keyring_worker_wedged:
        executor.shutdown(wait=True)

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 None when it is gone, else why it may survive.

"Gone" covers a successful delete and an entry that never existed — the caller's goal is absence, not the delete call. A locked keyring cannot prove absence and never prompts here, so it returns its reason; so does a backend that rejects the delete while the entry still reads back. Callers render the reason instead of guessing.

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

def forget_passphrase_in_keyring() -> str | None:
    """Remove the keyring entry; return ``None`` when it is gone, else why it may survive.

    "Gone" covers a successful delete and an entry that never existed —
    the caller's goal is absence, not the delete call.  A locked keyring
    cannot prove absence and never prompts here, so it returns its
    reason; so does a backend that rejects the delete while the entry
    still reads back.  Callers render the reason instead of guessing.
    """
    if (blocked := os_keyring_read_blocked()) is not None:
        return blocked
    try:
        import keyring  # noqa: PLC0415
        from keyring.errors import PasswordDeleteError  # noqa: PLC0415

        try:
            keyring.delete_password(KEYRING_SERVICE, KEYRING_USERNAME)
        except PasswordDeleteError:
            # Most backends raise this for a missing entry; only a
            # residual entry after it means the backend refused.
            if load_passphrase_from_keyring() is not None:
                return "the backend rejected the delete"
        return None
    except Exception as exc:  # noqa: BLE001
        return f"OS keyring unreachable ({type(exc).__name__})"

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_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_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)