"""Per-bucket instance-scoped unlock state.
`BucketSession` replaces the module-global `ClassVar` caches that
previously survived a bucket switch on `KeyringMasterKeyProvider` and
`FileFallbackMasterKeyProvider`. Each instance binds to exactly one
`bucket_id`; the unlocked KEK and DEK are held in `bytearray` buffers
so `close()` can overwrite the bytes in place before the references are
dropped. The session is the only object that holds cleartext key
material on the master-key surface; the substrate invariant
forbids any module-global mutable state that could survive a bucket
switch.
The `bytearray` zeroisation is best-effort. Python may have produced
short-lived `bytes` copies of the buffers when callers materialised the
`kek` / `dek` properties; the garbage collector owns the lifetime of
those copies. The contract is documented honestly so callers do not
assume Python guarantees a deeper wipe than the language can deliver.
See Also:
:class:`~adapters.persistence.storage.master_key.KeyringMasterKeyProvider`
Provider whose former process cache is replaced by this session.
:class:`~adapters.persistence.storage.master_key.FileFallbackMasterKeyProvider`
File-backed provider whose unlocked buffers are session-scoped.
``2026-05-14-profile-bucket-lifecycle-adr``
Decision that made unlocked key material bucket-session-owned.
"""
from __future__ import annotations
from datetime import datetime, timedelta
from typing import TYPE_CHECKING
from sqlalchemy.exc import SQLAlchemyError
from .....core.logging import get_logger
from ..bucket import BucketLockedError
from ..errors import (
storage_validation_error as _storage_validation_error,
)
from ._zeroise import zeroise as _zeroise
if TYPE_CHECKING:
from sqlalchemy import Engine
from .....core.config import Settings
_KEK_BYTES = 32
_DEK_BYTES = 32
_log = get_logger(__name__)
[docs]
class BucketSession:
"""One per-bucket unlock session.
The class deliberately holds NO `ClassVar` mutable state. Two
sessions for two distinct bucket ids own two independent
`bytearray` buffers and never share key material.
"""
__slots__ = (
"_bucket_id",
"_dek_buffer",
"_engine",
"_idle_deadline",
"_idle_window",
"_kek_buffer",
"_sealed",
"_unsecured_backend",
)
def __init__(
self,
*,
bucket_id: str,
kek_buffer: bytearray,
dek_buffer: bytearray,
idle_window: timedelta,
idle_deadline: datetime,
unsecured_backend: bool,
) -> None:
self._bucket_id = bucket_id
self._kek_buffer = kek_buffer
self._dek_buffer = dek_buffer
self._idle_window = idle_window
self._idle_deadline = idle_deadline
self._unsecured_backend = unsecured_backend
self._sealed = False
self._engine: Engine | None = None
[docs]
@classmethod
def open(
cls,
*,
bucket_id: str,
kek: bytes,
dek: bytes,
idle_minutes: int,
opened_at: datetime,
unsecured_backend: bool = False,
) -> BucketSession:
"""Open a session for one bucket.
Args:
bucket_id: Non-empty identifier of the bucket being unlocked.
kek: 32-byte Argon2id-derived key-encryption key.
dek: 32-byte data-encryption key recovered by unwrapping the
bucket's wrapped DEK under the KEK.
idle_minutes: Idle-timeout window in minutes; must be a
strict positive integer.
opened_at: UTC timestamp at which the session opened.
unsecured_backend: When ``True``, the session was opened
against an unsecured (non-OS-keychain) backend; callers
use this flag to emit appropriate warnings.
Returns:
A new :class:`BucketSession` with the provided credentials and TTL.
Raises:
StorageValidationError: When ``bucket_id`` is empty, ``idle_minutes`` is not
positive, ``kek`` is not 32 bytes, or ``dek`` is not 32 bytes.
"""
if not bucket_id:
raise _storage_validation_error("bucket_id must be non-empty")
if idle_minutes <= 0:
raise _storage_validation_error("idle_minutes must be a strict positive integer")
if len(kek) != _KEK_BYTES:
raise _storage_validation_error(f"kek must be exactly {_KEK_BYTES} bytes")
if len(dek) != _DEK_BYTES:
raise _storage_validation_error(f"dek must be exactly {_DEK_BYTES} bytes")
idle_window = timedelta(minutes=idle_minutes)
return cls(
bucket_id=bucket_id,
kek_buffer=bytearray(kek),
dek_buffer=bytearray(dek),
idle_window=idle_window,
idle_deadline=opened_at + idle_window,
unsecured_backend=unsecured_backend,
)
@property
def bucket_id(self) -> str:
return self._bucket_id
@property
def sealed(self) -> bool:
return self._sealed
@property
def unsecured_backend(self) -> bool:
return self._unsecured_backend
@property
def idle_deadline(self) -> datetime:
return self._idle_deadline
@property
def kek(self) -> bytes:
"""Return an immutable view of the live KEK bytes.
Raises `BucketLockedError` after `close()` has sealed the
session.
"""
if self._sealed:
raise BucketLockedError(bucket_id=self._bucket_id)
return bytes(self._kek_buffer)
@property
def dek(self) -> bytes:
"""Return an immutable view of the live DEK bytes.
Raises `BucketLockedError` after `close()` has sealed the
session.
"""
if self._sealed:
raise BucketLockedError(bucket_id=self._bucket_id)
return bytes(self._dek_buffer)
[docs]
def touch(self, now: datetime) -> None:
"""Reset the idle-timeout deadline to `now + idle_window`."""
if self._sealed:
raise BucketLockedError(bucket_id=self._bucket_id)
self._idle_deadline = now + self._idle_window
[docs]
def is_expired(self, now: datetime) -> bool:
"""Return whether the idle window has elapsed at `now`."""
if self._sealed:
return True
return now >= self._idle_deadline
[docs]
def acquire_engine(self, settings: Settings) -> Engine:
"""Lazily acquire and register this bucket's engine on first storage access.
The session is the single owner of the SQLAlchemy engine
lifecycle for its bucket: the first storage access within the
session resolves (or creates) the bucket engine and registers the
handle here, so :meth:`close` disposes exactly that engine on
session close or profile switch. Subsequent accesses return the
already-registered handle.
Args:
settings: The :class:`~core.config.Settings` routing to
this bucket's database, passed through to
:func:`~adapters.persistence.storage.sql.engine.get_engine`.
Returns:
The :class:`~sqlalchemy.engine.Engine` bound to this session.
Raises:
BucketLockedError: When the session has already been sealed.
"""
if self._sealed:
raise BucketLockedError(bucket_id=self._bucket_id)
if self._engine is None:
from ..sql.engine import get_engine
self._engine = get_engine(settings)
return self._engine
[docs]
def invalidate_engine(self) -> None:
"""Drop this session's cached engine handle without sealing the session.
:meth:`acquire_engine` caches its resolved handle for the life of
the session, so a caller that destroys and later re-materialises
this bucket's on-disk database out from under a still-open session
(the profile-reset / bucket-removal path) must invalidate the
session-level cache too, or the next :meth:`acquire_engine` call
returns the stale handle bound to a directory that no longer
exists. This is the session-scoped counterpart of
:func:`~adapters.persistence.storage.sql.engine.dispose_engines_for_bucket`,
which only evicts the process-wide engine cache; callers that
remove a bucket directory while its session may still be active
must call both. A no-op when no engine has been acquired yet or
the session is already sealed.
"""
if self._sealed or self._engine is None:
return
from ..sql.engine import dispose_engine_handle
engine = self._engine
self._engine = None
try:
dispose_engine_handle(engine)
except SQLAlchemyError as exc:
_log.debug("bucket session engine invalidation failed error_type=%s", type(exc).__name__)
[docs]
def close(self) -> None:
"""Zeroise key buffers, dispose the bucket's engine, and seal the session.
Idempotent: a second call after the first is a no-op.
Engine disposal:
The session owns the engine lifecycle for its bucket. Closing the
session (on idle expiry, profile switch, or explicit close)
disposes the engine handle registered at first storage access and
evicts every cached engine bound to this bucket id, so the next
consumer that opens a different bucket — or re-opens this one —
never reuses a stale engine handle. Disposal keys on bucket
identity, so it never depends on re-deriving a database route from
live settings.
"""
if self._sealed:
return
_zeroise(self._kek_buffer)
_zeroise(self._dek_buffer)
self._sealed = True
self._dispose_engine()
def _dispose_engine(self) -> None:
"""Dispose the engine(s) bound to this bucket's database."""
from ..sql.engine import dispose_engine_handle, dispose_engines_for_bucket
engine = self._engine
self._engine = None
try:
if engine is not None:
dispose_engine_handle(engine)
dispose_engines_for_bucket(self._bucket_id)
except SQLAlchemyError as exc:
_log.debug("bucket session engine disposal failed error_type=%s", type(exc).__name__)
__all__ = ["BucketSession"]