"""Encrypted Google record persistence.
This module writes Google records through
:class:`adapters.persistence.storage.SecureObjectRepository`.
Five per-profile record families back Google configuration and session state,
each under the namespace and
:class:`adapters.persistence.storage.SensitivityClass` declared by the
storage registry:
- :data:`adapters.persistence.storage.GOOGLE_OAUTH_CLIENT_NAMESPACE`
stores the operator-imported
:class:`adapters.outbound.google.OAuthClient` at ``SECRET``
sensitivity because ``client_secret`` is a long-lived credential.
- :data:`adapters.persistence.storage.GOOGLE_OAUTH_TOKEN_NAMESPACE`
stores the refresh :class:`adapters.outbound.google.OAuthToken`
returned by :func:`adapters.outbound.google.run_login_flow` at
``SECRET`` sensitivity.
- :data:`adapters.persistence.storage.GOOGLE_OAUTH_METADATA_NAMESPACE`
stores the non-secret
:class:`adapters.outbound.google.OAuthMetadata` account, scope,
issuance, refresh, and reauth audit fields at ``FINANCIAL`` sensitivity.
- :data:`adapters.persistence.storage.GOOGLE_DRIVE_CONFIG_NAMESPACE`
stores the :class:`adapters.outbound.google.DriveConfig` root folder
selection used by
:func:`adapters.outbound.storage.get_storage_provider` at
``FINANCIAL`` sensitivity.
- :data:`adapters.persistence.storage.GOOGLE_CREDENTIAL_SOURCE_NAMESPACE`
stores the :class:`adapters.outbound.google.GoogleCredentialSourceSelection`
choice of :class:`core.GoogleCredentialSourceKind` (and, for
service-account impersonation, the target SA email/scopes) at
``FINANCIAL`` sensitivity — configuration only, never a credential.
The public helpers use the profile identifier resolved by
:func:`adapters.outbound.google.resolve_active_profile` as the storage
object key, matching the ``{profile}`` grammar on all five namespace
definitions.
"""
from __future__ import annotations
from ....adapters.persistence.storage import (
GOOGLE_CREDENTIAL_SOURCE_NAMESPACE,
GOOGLE_DRIVE_CONFIG_NAMESPACE,
GOOGLE_OAUTH_CLIENT_NAMESPACE,
GOOGLE_OAUTH_METADATA_NAMESPACE,
GOOGLE_OAUTH_TOKEN_NAMESPACE,
SecureObjectRepository,
secure_object_repository_for_active_bucket,
)
from ....core.classification import SensitivityClass
from ....core.external_constants import UTF_8_ENCODING
from ....core.time import now
from ._impersonation import GoogleCredentialSourceSelection
from ._records import DriveConfig, OAuthClient, OAuthMetadata, OAuthToken
_NAMESPACE_CLIENT = GOOGLE_OAUTH_CLIENT_NAMESPACE.namespace
_NAMESPACE_TOKEN = GOOGLE_OAUTH_TOKEN_NAMESPACE.namespace
_NAMESPACE_METADATA = GOOGLE_OAUTH_METADATA_NAMESPACE.namespace
_NAMESPACE_DRIVE_CONFIG = GOOGLE_DRIVE_CONFIG_NAMESPACE.namespace
_NAMESPACE_CREDENTIAL_SOURCE = GOOGLE_CREDENTIAL_SOURCE_NAMESPACE.namespace
_RECORD_VERSION = 1
[docs]
def save_client(profile: str, client: OAuthClient) -> None:
"""Persist an :class:`adapters.outbound.google.OAuthClient` for ``profile``.
The record is written under
:data:`adapters.persistence.storage.GOOGLE_OAUTH_CLIENT_NAMESPACE`
with :class:`adapters.persistence.storage.SensitivityClass`
``SECRET`` so ``aeat config google login`` and Drive credential hydration
can reload the operator-imported Desktop OAuth client.
"""
_repository().save(
namespace=_NAMESPACE_CLIENT,
object_key=profile,
classification=SensitivityClass.SECRET,
schema_version=_RECORD_VERSION,
written_at=now(),
payload=client.model_dump_json().encode(UTF_8_ENCODING),
)
[docs]
def load_client(profile: str) -> OAuthClient | None:
"""Load the :class:`adapters.outbound.google.OAuthClient` for ``profile``.
Returns:
The stored :class:`adapters.outbound.google.OAuthClient`, or
``None`` when the profile has not registered a Desktop OAuth client.
"""
record = _repository().load(
_NAMESPACE_CLIENT,
profile,
expected_class=SensitivityClass.SECRET,
max_supported_version=_RECORD_VERSION,
)
if record is None:
return None
return OAuthClient.model_validate_json(record.payload.decode(UTF_8_ENCODING))
[docs]
def save_token(profile: str, token: OAuthToken) -> None:
"""Persist an :class:`adapters.outbound.google.OAuthToken` for ``profile``.
The token is written under
:data:`adapters.persistence.storage.GOOGLE_OAUTH_TOKEN_NAMESPACE`
with :class:`adapters.persistence.storage.SensitivityClass`
``SECRET``. The CLI saves this after
:func:`adapters.outbound.google.run_login_flow`, and
refresh code may overwrite it when Google rotates the refresh token.
"""
_repository().save(
namespace=_NAMESPACE_TOKEN,
object_key=profile,
classification=SensitivityClass.SECRET,
schema_version=_RECORD_VERSION,
written_at=now(),
payload=token.model_dump_json().encode(UTF_8_ENCODING),
)
[docs]
def load_token(profile: str) -> OAuthToken | None:
"""Load the :class:`adapters.outbound.google.OAuthToken` for ``profile``.
Returns:
The stored :class:`adapters.outbound.google.OAuthToken`, or
``None`` when the profile has no active Google login session.
"""
record = _repository().load(
_NAMESPACE_TOKEN,
profile,
expected_class=SensitivityClass.SECRET,
max_supported_version=_RECORD_VERSION,
)
if record is None:
return None
return OAuthToken.model_validate_json(record.payload.decode(UTF_8_ENCODING))
[docs]
def save_drive_config(profile: str, config: DriveConfig) -> None:
"""Persist the per-profile :class:`adapters.outbound.google.DriveConfig`.
The config is written under
:data:`adapters.persistence.storage.GOOGLE_DRIVE_CONFIG_NAMESPACE`
with :class:`adapters.persistence.storage.SensitivityClass`
``FINANCIAL`` so
:func:`adapters.outbound.storage.get_storage_provider` can resolve
the Drive root folder without re-reading environment-only configuration.
"""
_repository().save(
namespace=_NAMESPACE_DRIVE_CONFIG,
object_key=profile,
classification=SensitivityClass.FINANCIAL,
schema_version=_RECORD_VERSION,
written_at=now(),
payload=config.model_dump_json().encode(UTF_8_ENCODING),
)
[docs]
def load_drive_config(profile: str) -> DriveConfig | None:
"""Load the per-profile :class:`adapters.outbound.google.DriveConfig`.
Returns:
The stored :class:`adapters.outbound.google.DriveConfig`, or
``None`` when the profile has no persisted Drive root folder selection.
"""
record = _repository().load(
_NAMESPACE_DRIVE_CONFIG,
profile,
expected_class=SensitivityClass.FINANCIAL,
max_supported_version=_RECORD_VERSION,
)
if record is None:
return None
return DriveConfig.model_validate_json(record.payload.decode(UTF_8_ENCODING))
[docs]
def save_credential_source_selection(profile: str, selection: GoogleCredentialSourceSelection) -> None:
"""Persist the per-profile :class:`adapters.outbound.google.GoogleCredentialSourceSelection`.
The record is written under
:data:`adapters.persistence.storage.GOOGLE_CREDENTIAL_SOURCE_NAMESPACE`
with :class:`adapters.persistence.storage.SensitivityClass`
``FINANCIAL`` so
:func:`adapters.outbound.storage.build_google_credentials` can dispatch
to the chosen :class:`core.GoogleCredentialSourceKind` without
re-reading environment-only configuration. No long-lived secret rides
on this record: the impersonated access token is re-derived from
Application Default Credentials on every use and is never persisted.
"""
_repository().save(
namespace=_NAMESPACE_CREDENTIAL_SOURCE,
object_key=profile,
classification=SensitivityClass.FINANCIAL,
schema_version=_RECORD_VERSION,
written_at=now(),
payload=selection.model_dump_json().encode(UTF_8_ENCODING),
)
[docs]
def load_credential_source_selection(profile: str) -> GoogleCredentialSourceSelection | None:
"""Load the per-profile :class:`adapters.outbound.google.GoogleCredentialSourceSelection`.
Returns:
The stored :class:`adapters.outbound.google.GoogleCredentialSourceSelection`,
or ``None`` when the profile has no persisted selection. A ``None``
result means the default
:attr:`core.GoogleCredentialSourceKind.OAUTH_DESKTOP` path applies —
callers must not treat a missing record as an error.
"""
record = _repository().load(
_NAMESPACE_CREDENTIAL_SOURCE,
profile,
expected_class=SensitivityClass.FINANCIAL,
max_supported_version=_RECORD_VERSION,
)
if record is None:
return None
return GoogleCredentialSourceSelection.model_validate_json(record.payload.decode(UTF_8_ENCODING))
[docs]
def delete_session(profile: str) -> tuple[bool, bool]:
"""Delete the login session while preserving registration and Drive config.
Removes only the
:data:`adapters.persistence.storage.GOOGLE_OAUTH_TOKEN_NAMESPACE` and
:data:`adapters.persistence.storage.GOOGLE_OAUTH_METADATA_NAMESPACE`
records, matching ``aeat config google logout``. The registered
:class:`adapters.outbound.google.OAuthClient` and
:class:`adapters.outbound.google.DriveConfig` remain available so a
later login can reuse the Cloud Console JSON and the same Drive root
folder.
Args:
profile: The profile identifier whose token and metadata records to
delete.
Returns:
A pair ``(token_removed, metadata_removed)``.
"""
repo = _repository()
token_removed = repo.delete(_NAMESPACE_TOKEN, profile)
metadata_removed = repo.delete(_NAMESPACE_METADATA, profile)
return token_removed, metadata_removed
__all__ = [
"delete_session",
"load_client",
"load_credential_source_selection",
"load_drive_config",
"load_metadata",
"load_token",
"save_client",
"save_credential_source_selection",
"save_drive_config",
"save_metadata",
"save_token",
]
def _repository() -> SecureObjectRepository:
return secure_object_repository_for_active_bucket()