"""Secure-object repair diagnostics, remediation decisions, and policy coverage.
The read side of this module produces metadata-only reports over encrypted
secure-object rows. Integrity sweeps iterate namespaces in a
:class:`~aeat.adapters.persistence.storage.SecureObjectRepository` and return
per-namespace
:class:`~aeat.adapters.persistence.storage.SecureObjectNamespaceIntegrity`
counts; inventory rows expose row metadata and HMAC digests, never natural
keys or payload bytes. These reports
back the repair integrity surface and the quarantine dry-run path without
emitting bucket events.
The write side is deliberately narrow: :class:`RepairRemediationDecision`
records persist non-destructive preserve / quarantine / rebuild /
export-required planning outcomes as encrypted AUDIT-class rows. A decision
record is evidence for a later operator workflow, not mutation authority.
The policy catalog returned by
:func:`build_repair_policy_command_surface_catalog` mirrors repair, recovery,
import, export, and bucket-history command surfaces against registered
:class:`SecureObjectNamespaceDefinition` metadata. Tests use it as a drift gate
so new maintenance surfaces cannot appear without an explicit namespace policy.
See Also:
:mod:`aeat.application.diagnostics`
Builds the user-facing repair report and delegates quarantine preview /
commit flows through this module's active-bucket repair session.
:class:`~aeat.adapters.persistence.storage.SecureObjectRepository`
Encrypted SQL repository whose namespace integrity probes and
quarantine operation supply the repair data.
:data:`~aeat.adapters.persistence.storage.STORAGE_NAMESPACE_REGISTRY`
Central registry copied into repair-policy namespace rows.
:mod:`aeat.entrypoints.cli._config._repair_cli`
CLI command surface that renders these reports and policy-backed repair
actions.
"""
from __future__ import annotations
from collections.abc import Generator, Iterator, Sequence
from contextlib import contextmanager
from datetime import datetime
from typing import Literal, Protocol, runtime_checkable
from pydantic import BaseModel, Field
from ..adapters.persistence.storage import (
REPAIR_INTEGRITY_DECISION_NAMESPACE as REPAIR_DECISION_STORAGE_NAMESPACE,
)
from ..adapters.persistence.storage import (
STORAGE_NAMESPACE_REGISTRY,
WORKFLOW_STATE_NAMESPACE,
SecureObjectNamespaceDefinition,
SecureObjectNamespaceIntegrity,
SecureObjectRepository,
)
from ..adapters.persistence.storage.sql.secure_objects import (
SecureObjectDecryptabilityRow,
)
from ..core import STRICT_FROZEN_CONFIG
from ..core.errors import CoreError
from ..core.hashing import content_hash_hex
from ..core.logging import get_logger
from .diagnostics import DiagnosticCheck
_log = get_logger(__name__)
[docs]
class RepairIntegrityError(CoreError):
"""Base error for secure-object integrity and repair-remediation failures.
Raised by the repair-integrity application layer when an integrity
invariant or remediation contract is violated. Inherits from
:class:`aeat.core.errors.CoreError` so callers can catch either the
specific subclass or the broad domain base without importing the full
repair-integrity module.
"""
[docs]
class RepairDecisionNotFoundError(RepairIntegrityError):
"""Raised when a repair-remediation decision lookup misses its target.
Fired by :meth:`RepairRemediationDecisionRepository.load_decision`
when no row exists for the requested ``decision_id``. Distinct from
a generic not-found so the CLI can produce an actionable hint
(e.g. list existing decisions) without catching broad exception classes.
"""
_REPAIR_DECISION_NAMESPACE = REPAIR_DECISION_STORAGE_NAMESPACE.namespace
"""Profile-local secure-object namespace for repair-remediation decisions.
The encrypted secure-object rows live in the active profile's bucket;
cross-profile reads are not permitted by the underlying repository.
"""
_RepairDecisionOutcome = Literal["preserve", "quarantine", "rebuild", "export-required"]
"""Closed set of non-destructive planning outcomes for repair-remediation."""
@runtime_checkable
class _SecureObjectRepositoryProtocol(Protocol):
"""Structural interface consumed by the repair-integrity application layer.
:class:`~aeat.adapters.persistence.storage.SecureObjectRepository`
satisfies this protocol. The interface is limited to namespace enumeration,
integrity probing, key inventory, and
:class:`~aeat.adapters.persistence.storage.sql.secure_objects.SecureObjectDecryptabilityRow`
iteration so read-only repair reports can be tested against the real
repository without granting mutation APIs to the report builders.
"""
def list_namespaces(self) -> tuple[str, ...]: ...
def probe_namespace_integrity(self, namespace: str) -> SecureObjectNamespaceIntegrity: ...
def list_keys(self, namespace: str) -> tuple[str, ...]: ...
def iter_namespace_decryptability(self, namespace: str) -> Iterator[SecureObjectDecryptabilityRow]: ...
[docs]
class RepairIntegrityReport(BaseModel):
"""Metadata-only secure-object integrity report.
``namespaces`` carries one :class:`SecureObjectNamespaceIntegrity` per
probed namespace. ``check`` is the aggregate :class:`DiagnosticCheck` row
used by repair renderers; failing reports carry the quarantine command as
the next action.
See Also:
:func:`build_repair_integrity_report`
Producer that fills this report from real secure-object
decryptability probes.
:class:`~aeat.application.diagnostics.SecureObjectIntegrityReport`
Config-repair rollup that uses the same namespace integrity shape.
"""
model_config = STRICT_FROZEN_CONFIG
namespaces: tuple[SecureObjectNamespaceIntegrity, ...]
readable_total: int = Field(ge=0)
unreadable_total: int = Field(ge=0)
check: DiagnosticCheck
[docs]
class RepairListRow(BaseModel):
"""One metadata row in the secure-object repair inventory.
Rows are projected from
:class:`~aeat.adapters.persistence.storage.sql.secure_objects.SecureObjectDecryptabilityRow`
values returned by
:meth:`~aeat.adapters.persistence.storage.SecureObjectRepository.iter_namespace_decryptability`.
``object_key_digest`` is the stored HMAC digest, not the natural object key.
``reason`` is populated only for unreadable rows and must remain a diagnostic
class of failure rather than decrypted payload context.
"""
model_config = STRICT_FROZEN_CONFIG
namespace: str = Field(min_length=1)
object_key_digest: str = Field(min_length=1)
readable: bool | None = None
row_id: int | None = Field(default=None, ge=0)
classification: str | None = None
schema_version: int | None = Field(default=None, ge=1)
written_at: datetime | None = None
reason: str | None = None
[docs]
class RepairListReport(BaseModel):
"""Internal secure-object inventory report for one namespace.
The report combines the namespace's :class:`SecureObjectNamespaceIntegrity`
counts with metadata-only :class:`RepairListRow` entries. Operator-facing
repair commands render aggregate integrity and quarantine previews rather
than exposing a broad raw list command.
"""
model_config = STRICT_FROZEN_CONFIG
namespace: str = Field(min_length=1)
integrity: SecureObjectNamespaceIntegrity
rows: tuple[RepairListRow, ...]
rows_total: int = Field(ge=0)
filter_mode: str = Field(min_length=1)
def _aggregate_integrity(
integrity: tuple[SecureObjectNamespaceIntegrity, ...],
) -> DiagnosticCheck:
"""Render the cross-namespace summary as one :class:`DiagnosticCheck` row.
The check honours the exhaustiveness lock: ``fail`` / ``warn`` rows
MUST carry exactly one of ``next_action`` / ``dead_end``; ``ok`` rows
MUST carry neither.
"""
readable = sum(item.readable for item in integrity)
unreadable = sum(item.unreadable for item in integrity)
if unreadable == 0:
return DiagnosticCheck(
name="secure_objects.integrity",
status="ok",
summary=(f"{readable} row(s) decryptable across {len(integrity)} namespace(s)"),
)
impacted = ", ".join(
f"{item.namespace} ({item.unreadable}/{item.readable + item.unreadable})"
for item in integrity
if item.unreadable
)
return DiagnosticCheck(
name="secure_objects.integrity",
status="fail",
summary=f"{unreadable} undecryptable row(s) in: {impacted}",
next_action="aeat config repair quarantine --yes",
)
[docs]
def build_repair_integrity_report(
*,
namespace: str | None = None,
repository: _SecureObjectRepositoryProtocol | None = None,
) -> RepairIntegrityReport:
"""Probe secure-object decryptability and return a :class:`RepairIntegrityReport`.
When ``namespace`` is set, only that namespace is probed; otherwise every
namespace currently present in the repository is scanned. Without an injected
repository, the probe opens the active bucket repair session before resolving
the runtime-bound :class:`SecureObjectRepository`.
"""
if repository is not None:
return _build_repair_integrity_report(namespace=namespace, repository=repository)
with active_bucket_repair_session():
return _build_repair_integrity_report(namespace=namespace, repository=_active_bucket_repair_repository())
def _build_repair_integrity_report(
*,
namespace: str | None,
repository: _SecureObjectRepositoryProtocol,
) -> RepairIntegrityReport:
repo = repository
namespaces = repo.list_namespaces() if namespace is None else (namespace,)
integrity = tuple(repo.probe_namespace_integrity(ns) for ns in namespaces)
readable_total = sum(item.readable for item in integrity)
unreadable_total = sum(item.unreadable for item in integrity)
return RepairIntegrityReport(
namespaces=integrity,
readable_total=readable_total,
unreadable_total=unreadable_total,
check=_aggregate_integrity(integrity),
)
[docs]
@contextmanager
def active_bucket_repair_session() -> Generator[None]:
"""Best-effort session opener for active-bucket repair diagnostics.
Repair integrity and quarantine previews need to test decryptability under
the active bucket key, but they are bootstrap-adjacent diagnostics rather
than profile enrollment flows. If a session is already active it is reused;
otherwise the configured master-key provider is entered for the span. A
provider failure is logged and the caller still observes the normal
repository/runtime readiness result.
See Also:
:func:`~aeat.application.diagnostics.preview_quarantine_unreadable_secure_objects`
Dry-run repair flow that uses this context before probing
decryptability.
:func:`~aeat.application.diagnostics.quarantine_unreadable_secure_objects`
Commit flow that uses this context before calling
:meth:`~aeat.adapters.persistence.storage.SecureObjectRepository.quarantine_unreadable_rows`.
"""
provider: object | None = None
try:
from ..adapters.persistence.storage import get_master_key_provider, has_active_bucket_session
if has_active_bucket_session():
yield
return
provider = get_master_key_provider()
# TYPE-IGNORE-RATIONALE-RUNTIME-CM-PROTOCOL:
# get_master_key_provider returns object; __enter__/__exit__ are correct
# at runtime but unverifiable without a typed Protocol.
provider.__enter__() # type: ignore[attr-defined]
except Exception as exc:
_log.debug("repair integrity could not open active bucket session: %s", type(exc).__name__)
yield
return
try:
yield
finally:
# TYPE-IGNORE-RATIONALE-RUNTIME-CM-PROTOCOL:
# get_master_key_provider returns object; __enter__/__exit__ are correct
# at runtime but unverifiable without a typed Protocol.
provider.__exit__(None, None, None) # type: ignore[attr-defined]
[docs]
def build_repair_list_report(
*,
namespace: str,
include_all: bool = False,
only_unreadable: bool = False,
repository: _SecureObjectRepositoryProtocol | None = None,
) -> RepairListReport:
"""List secure-object metadata stored under ``namespace``.
``--all`` returns every key; ``--unreadable`` filters to only the
rows whose payload cannot be decrypted under the current master
key. Default behaviour (both flags False) returns the full key set
but caps the inventory at the integrity-readable count for
bandwidth control on large namespaces — same as ``--all`` for
namespaces with no integrity issues.
Returns a :class:`RepairListReport` enumerating matching HMAC digests and
their decryptability status. The report never exposes natural object keys or
payload bytes. Without an injected repository, the list path enters
:func:`active_bucket_repair_session` before resolving the active
:class:`SecureObjectRepository`.
"""
if include_all and only_unreadable:
raise RepairIntegrityError(
translated_message="application.repair_integrity.errors.conflicting_list_filters",
context={"filters": "--all,--unreadable"},
)
if repository is None:
with active_bucket_repair_session():
return _build_repair_list_report(
namespace=namespace,
include_all=include_all,
only_unreadable=only_unreadable,
repository=_active_bucket_repair_repository(),
)
return _build_repair_list_report(
namespace=namespace,
include_all=include_all,
only_unreadable=only_unreadable,
repository=repository,
)
def _build_repair_list_report(
*,
namespace: str,
include_all: bool,
only_unreadable: bool,
repository: _SecureObjectRepositoryProtocol,
) -> RepairListReport:
repo = repository
integrity = repo.probe_namespace_integrity(namespace)
row_metadata = tuple(repo.iter_namespace_decryptability(namespace))
rows = tuple(_repair_list_row(row) for row in row_metadata)
if only_unreadable:
rows = tuple(row for row in rows if row.readable is False)
filter_mode = "unreadable"
elif include_all:
filter_mode = "all"
else:
filter_mode = "default"
return RepairListReport(
namespace=namespace,
integrity=integrity,
rows=rows,
rows_total=len(rows),
filter_mode=filter_mode,
)
def _active_bucket_repair_repository() -> SecureObjectRepository:
from ..adapters.persistence.storage import secure_object_repository_for_active_bucket
return secure_object_repository_for_active_bucket()
def _repair_list_row(row: SecureObjectDecryptabilityRow) -> RepairListRow:
return RepairListRow(
namespace=row.namespace,
object_key_digest=row.object_key.hex(),
readable=row.readable,
row_id=row.row_id,
classification=row.classification,
schema_version=row.schema_version,
written_at=row.written_at,
reason=row.reason if row.readable is False else None,
)
def _expected_repair_decision_id(decision: RepairRemediationDecision) -> str:
return repair_remediation_decision_id(
target_namespace=decision.target_namespace,
target_object_key_digest=decision.target_object_key_digest,
outcome=decision.outcome,
decided_at=decision.decided_at,
decided_by=decision.decided_by,
reason=decision.reason,
likely_origin=decision.likely_origin,
replacement_evidence_requirements=decision.replacement_evidence_requirements,
verified_replacement_evidence_refs=decision.verified_replacement_evidence_refs,
)
[docs]
class RepairPolicyNamespaceClassification(BaseModel):
"""Minimal namespace classification attached to a repair-policy surface.
Embedded in :class:`RepairPolicyNamespacePolicy` so command catalog rows can
describe non-registered bundle / filing / ledger surfaces and registered
secure-object namespace scopes with the same shape.
"""
model_config = STRICT_FROZEN_CONFIG
role: str = Field(min_length=1)
[docs]
class RepairPolicyNamespacePolicy(BaseModel):
"""Policy metadata for one namespace governed by a command surface.
Registered secure-object namespaces copy owner, sensitivity, schema version,
and scope from :class:`SecureObjectNamespaceDefinition` so repair and
recovery surfaces stay tied to the namespace registry instead of parallel
role markers. :class:`RepairPolicyCommandSurface` attaches these rows to
every CLI surface that can inspect, repair, import, export, or recover
namespace-owned data.
"""
model_config = STRICT_FROZEN_CONFIG
namespace_classification: RepairPolicyNamespaceClassification
owner_domain: str = Field(min_length=1)
repair_policy: str = Field(min_length=1)
recovery_policy: str = Field(min_length=1)
mutation_authority: str = Field(min_length=1)
registered_namespace_key: str | None = Field(default=None, min_length=1)
registered_namespace: str | None = Field(default=None, min_length=1)
registered_owner: str | None = Field(default=None, min_length=1)
registered_sensitivity: str | None = Field(default=None, min_length=1)
registered_schema_version: int | None = Field(default=None, ge=1)
registered_scope: str | None = Field(default=None, min_length=1)
[docs]
class RepairPolicyCommandSurface(BaseModel):
"""One catalogued repair-policy CLI command surface.
Each row links a command path to its owner domains, decision-trail anchors,
and the namespace policies that constrain any repair or recovery behavior
reachable from that command family.
See Also:
:class:`RepairPolicyNamespacePolicy`
Per-namespace policy rows attached to a command surface.
:func:`build_repair_policy_command_surface_catalog`
Executable catalog that mirrors the repair and recovery CLI surface.
"""
model_config = STRICT_FROZEN_CONFIG
command_path: str = Field(min_length=1)
"""The canonical CLI command path (e.g. ``config repair integrity objects``)."""
command_family: str = Field(min_length=1)
owner_domains: tuple[str, ...]
decision_links: tuple[str, ...]
namespace_policies: tuple[RepairPolicyNamespacePolicy, ...] = ()
_PROFILE_BUNDLE_POLICY = RepairPolicyNamespacePolicy(
namespace_classification=RepairPolicyNamespaceClassification(role="profile_bundle"),
owner_domain="profile_lifecycle",
repair_policy="preserve_manifest_and_encrypted_payload_identity",
recovery_policy="profile_import_export_roundtrip",
mutation_authority="explicit_operator_confirmation_required_for_profile_mutation",
)
_MODEL_FILING_POLICY = RepairPolicyNamespacePolicy(
namespace_classification=RepairPolicyNamespaceClassification(role="modelo_filing_artifact"),
owner_domain="modelo_filing",
repair_policy="preserve_exported_or_imported_filing_evidence",
recovery_policy="reimport_authoritative_filing_evidence",
mutation_authority="operator_requested_import_or_export_only",
)
_LEDGER_POLICY = RepairPolicyNamespacePolicy(
namespace_classification=RepairPolicyNamespaceClassification(role="ledger_artifact"),
owner_domain="ledger",
repair_policy="preserve_transaction_evidence_and_import_payloads",
recovery_policy="reimport_authoritative_ledger_source",
mutation_authority="operator_requested_import_or_export_only",
)
_REPAIR_DECISION_LINKS: tuple[str, ...] = (
# Catalog itself (build_repair_policy_command_surface_catalog,
# repair_remediation_decision_id, and the linkage-coordination
# contract these surfaces serve as the executable mirror of).
"[[2026-05-26-linkage-design-audit]]",
# Repair CLI shape (the operator-facing verb surface every entry
# in the catalog mirrors — adding a verb here without an entry
# there is the drift the gate exists to catch).
"[[2026-05-13-cli-workflow-redesign-config-repair-shape]]",
)
def _secure_object_policy(
definition: SecureObjectNamespaceDefinition,
*,
repair_policy: str = "metadata_only_digest_inventory_preserve_ciphertext",
recovery_policy: str = "restore_matching_master_key_or_rebuild_from_authoritative_source",
mutation_authority: str = "explicit_operator_confirmation_required_for_mutation",
) -> RepairPolicyNamespacePolicy:
return RepairPolicyNamespacePolicy(
namespace_classification=RepairPolicyNamespaceClassification(role=definition.scope.value),
owner_domain=definition.owner,
repair_policy=repair_policy,
recovery_policy=recovery_policy,
mutation_authority=mutation_authority,
registered_namespace_key=definition.key,
registered_namespace=definition.namespace,
registered_owner=definition.owner,
registered_sensitivity=definition.sensitivity.value,
registered_schema_version=definition.schema_version,
registered_scope=definition.scope.value,
)
def _all_secure_object_namespace_policies() -> tuple[RepairPolicyNamespacePolicy, ...]:
return tuple(_secure_object_policy(definition) for definition in STORAGE_NAMESPACE_REGISTRY.namespaces)
def _surface(
command_path: str,
*,
command_family: str,
owner_domains: tuple[str, ...],
namespace_policies: tuple[RepairPolicyNamespacePolicy, ...] = (),
) -> RepairPolicyCommandSurface:
return RepairPolicyCommandSurface(
command_path=command_path,
command_family=command_family,
owner_domains=owner_domains,
decision_links=_REPAIR_DECISION_LINKS,
namespace_policies=namespace_policies,
)
[docs]
def build_repair_policy_command_surface_catalog() -> tuple[RepairPolicyCommandSurface, ...]:
"""Return the :class:`RepairPolicyCommandSurface` catalog for repair-policy CLI surfaces.
The catalog mirrors the Typer command registry for repair,
recovery, import, export, and bucket-history surfaces. It is used
as an executable drift gate: adding a new command in those families requires
a policy row here, and secure-object rows must derive their metadata from the
central namespace registry. Each row also carries decision links for
:class:`RepairRemediationDecision` governance.
See Also:
:data:`~aeat.adapters.persistence.storage.STORAGE_NAMESPACE_REGISTRY`
Source of secure-object namespace metadata copied into catalog
policies.
:class:`~aeat.adapters.persistence.storage.SecureObjectNamespaceDefinition`
Registered namespace declaration projected into policy rows.
"""
return (
_surface("config repair logs", command_family="repair", owner_domains=("diagnostics",)),
_surface(
"config repair quarantine",
command_family="repair",
owner_domains=("secure_storage",),
namespace_policies=_all_secure_object_namespace_policies(),
),
_surface(
"config repair reset-progress",
command_family="repair",
owner_domains=("workflow_state",),
namespace_policies=(
_secure_object_policy(
WORKFLOW_STATE_NAMESPACE,
repair_policy="metadata_only_workflow_state_reset_plan",
recovery_policy="restore_matching_master_key_or_rebuild_workflow_state_projection",
),
),
),
_surface(
"config repair profile",
command_family="repair",
owner_domains=("profile_lifecycle",),
namespace_policies=(_PROFILE_BUNDLE_POLICY,),
),
_surface(
"config repair integrity objects",
command_family="repair",
owner_domains=("secure_storage",),
namespace_policies=_all_secure_object_namespace_policies(),
),
_surface("config repair integrity registry", command_family="repair", owner_domains=("registry",)),
_surface("config repair connectivity", command_family="repair", owner_domains=("remote_connectivity",)),
_surface(
"config profile import",
command_family="recovery",
owner_domains=("profile_lifecycle",),
namespace_policies=(_PROFILE_BUNDLE_POLICY,),
),
_surface(
"config profile export",
command_family="recovery",
owner_domains=("profile_lifecycle",),
namespace_policies=(_PROFILE_BUNDLE_POLICY,),
),
_surface(
"config recover",
command_family="recovery",
owner_domains=("secure_storage",),
),
_surface(
"config rekey",
command_family="recovery",
owner_domains=("secure_storage",),
),
_surface(
"config show-recovery",
command_family="recovery",
owner_domains=("secure_storage",),
),
_surface(
"config verify-recovery",
command_family="recovery",
owner_domains=("secure_storage",),
),
_surface("config profile history", command_family="audit", owner_domains=("bucket_lifecycle",)),
_surface(
"app ledger import",
command_family="recovery",
owner_domains=("ledger",),
namespace_policies=(_LEDGER_POLICY,),
),
_surface(
"app ledger export",
command_family="recovery",
owner_domains=("ledger",),
namespace_policies=(_LEDGER_POLICY,),
),
_surface(
"app modelo export",
command_family="recovery",
owner_domains=("modelo_filing",),
namespace_policies=(_MODEL_FILING_POLICY,),
),
_surface(
"app modelo filing-record import",
command_family="recovery",
owner_domains=("modelo_filing",),
namespace_policies=(_MODEL_FILING_POLICY,),
),
_surface(
"app modelo audit export",
command_family="audit",
owner_domains=("modelo_audit", "modelo_filing"),
namespace_policies=(_MODEL_FILING_POLICY,),
),
)
__all__ = [
"RepairIntegrityReport",
"RepairListReport",
"RepairListRow",
"RepairPolicyCommandSurface",
"RepairPolicyNamespaceClassification",
"RepairPolicyNamespacePolicy",
"RepairRemediationDecision",
"RepairRemediationDecisionRepository",
"build_repair_integrity_report",
"build_repair_list_report",
"build_repair_policy_command_surface_catalog",
"repair_remediation_decision_id",
]