Source code for aeat.adapters.persistence.storage.sql._secure_object_integrity

"""Decryptability diagnostics for SQL secure-object persistence."""

from __future__ import annotations

from collections.abc import Iterator
from logging import Logger

from sqlalchemy import Engine, bindparam, text

from .....core.external_constants import UTF_8_ENCODING
from ..crypto import (
    decrypt_secure_object_payload,
    secure_object_payload_aad,
)
from ..errors import DecryptionError
from . import _orm
from ._secure_object_records import SecureObjectDecryptabilityRow, SecureObjectNamespaceIntegrity
from ._secure_object_schema import database_bytes, ensure_quarantine_table, quarantine_timestamp
from .session import session_scope


[docs] def quarantine_unreadable_rows( engine: Engine, *, logger: Logger, ) -> tuple[SecureObjectNamespaceIntegrity, ...]: """Move every undecryptable row into ``secure_objects_quarantine``. Returns a tuple of :class:`SecureObjectNamespaceIntegrity` records, one per namespace. """ ensure_quarantine_table(engine) with session_scope(engine) as session: quarantined_at = quarantine_timestamp() namespaces = ( session.execute(text("SELECT DISTINCT namespace FROM secure_objects ORDER BY namespace")).scalars().all() ) per_namespace: list[SecureObjectNamespaceIntegrity] = [] for namespace in namespaces: rows = session.execute( text( "SELECT id, object_key, classification, schema_version, written_at, " "revision_id, previous_revision_id, revision_ancestor_ids, " "previous_payload_hash, payload_hash, " "ciphertext_hash, revision_written_at, write_provenance, source_event_id, " "conflict_policy, payload " "FROM secure_objects WHERE namespace = :namespace", ).bindparams(bindparam("namespace", value=namespace)), ).all() quarantined = 0 retained = 0 for raw in rows: payload_bytes = raw.payload if isinstance(raw.payload, bytes) else bytes(raw.payload) object_key_value = ( raw.object_key if isinstance(raw.object_key, bytes | bytearray | memoryview) else str(raw.object_key).encode(UTF_8_ENCODING) ) object_key_bytes = bytes(object_key_value) try: decrypt_secure_object_payload( payload_bytes, associated_data=secure_object_payload_aad( namespace, object_key_bytes, int(raw.schema_version), ), ) except DecryptionError as exc: logger.debug( "secure_objects: quarantining unreadable row id=%s namespace=%s (%s)", int(raw.id), namespace, exc, ) session.execute( text( "INSERT INTO secure_objects_quarantine " "(source_id, namespace, object_key, classification, schema_version, " " written_at, revision_id, previous_revision_id, previous_payload_hash, " " revision_ancestor_ids, payload_hash, ciphertext_hash, " " revision_written_at, write_provenance, " " source_event_id, conflict_policy, payload, quarantined_at) " "VALUES (:source_id, :namespace, :object_key, :classification, " " :schema_version, :written_at, :revision_id, " " :previous_revision_id, :previous_payload_hash, " " :revision_ancestor_ids, :payload_hash, " " :ciphertext_hash, :revision_written_at, :write_provenance, " " :source_event_id, :conflict_policy, :payload, :quarantined_at)", ), { "source_id": int(raw.id), "namespace": namespace, "object_key": object_key_bytes, "classification": str(raw.classification), "schema_version": int(raw.schema_version), "written_at": raw.written_at, "revision_id": raw.revision_id, "previous_revision_id": raw.previous_revision_id, "revision_ancestor_ids": raw.revision_ancestor_ids, "previous_payload_hash": raw.previous_payload_hash, "payload_hash": raw.payload_hash, "ciphertext_hash": raw.ciphertext_hash, "revision_written_at": raw.revision_written_at, "write_provenance": raw.write_provenance, "source_event_id": raw.source_event_id, "conflict_policy": raw.conflict_policy, "payload": payload_bytes, "quarantined_at": quarantined_at, }, ) session.execute( text("DELETE FROM secure_objects WHERE id = :id"), {"id": int(raw.id)}, ) quarantined += 1 else: retained += 1 per_namespace.append( SecureObjectNamespaceIntegrity( namespace=namespace, readable=retained, unreadable=quarantined, ), ) return tuple(per_namespace)
[docs] def probe_namespace_integrity( engine: Engine, namespace: str, *, logger: Logger, ) -> SecureObjectNamespaceIntegrity: """Count decryptable and undecryptable rows in ``namespace``. Returns a :class:`SecureObjectNamespaceIntegrity` for the given namespace. """ readable = 0 unreadable = 0 with session_scope(engine) as session: stmt = text( "SELECT object_key, schema_version, payload FROM secure_objects WHERE namespace = :namespace", ).bindparams( bindparam("namespace", value=namespace), ) rows = session.execute(stmt).all() for raw in rows: try: decrypt_secure_object_payload( bytes(raw.payload), associated_data=secure_object_payload_aad( namespace, bytes(raw.object_key), int(raw.schema_version), ), ) except DecryptionError as exc: logger.debug( "secure_objects probe: unreadable row in namespace=%s (%s)", namespace, exc, ) unreadable += 1 else: readable += 1 return SecureObjectNamespaceIntegrity( namespace=namespace, readable=readable, unreadable=unreadable, )
[docs] def iter_namespace_decryptability( engine: Engine, namespace: str, ) -> Iterator[SecureObjectDecryptabilityRow]: """Yield :class:`SecureObjectDecryptabilityRow` records for one namespace.""" with session_scope(engine) as session: stmt = ( text( "SELECT id, object_key, classification, schema_version, written_at, payload " "FROM secure_objects WHERE namespace = :namespace " "ORDER BY object_key", ) .bindparams(bindparam("namespace", value=namespace)) .columns( id=_orm.SecureObjectRow.__table__.c.id.type, object_key=_orm.SecureObjectRow.__table__.c.object_key.type, classification=_orm.SecureObjectRow.__table__.c.classification.type, schema_version=_orm.SecureObjectRow.__table__.c.schema_version.type, written_at=_orm.SecureObjectRow.__table__.c.written_at.type, ) ) rows = session.execute(stmt).all() for raw in rows: object_key_value = database_bytes(raw.object_key) payload_value = database_bytes(raw.payload) try: decrypt_secure_object_payload( payload_value, associated_data=secure_object_payload_aad( namespace, object_key_value, int(raw.schema_version), ), ) except DecryptionError as exc: yield SecureObjectDecryptabilityRow( namespace=namespace, row_id=int(raw.id), object_key=object_key_value, classification=str(raw.classification), schema_version=int(raw.schema_version), written_at=raw.written_at, readable=False, reason=str(exc), ) else: yield SecureObjectDecryptabilityRow( namespace=namespace, row_id=int(raw.id), object_key=object_key_value, classification=str(raw.classification), schema_version=int(raw.schema_version), written_at=raw.written_at, readable=True, )