ss-tools/scripts/dr_evidence_scope.py

# #region Tooling.DrEvidenceCheck.TypedScope [C:4] [TYPE Module] [SEMANTICS dr,evidence,scope]
# @BRIEF Extract only schema-validated catalog, authority and active retained artifact references.
import json
import os

# #region Tooling.DrEvidenceCheck.TypedScope.Scopecontractrefused [C:3] [TYPE Class] [SEMANTICS dr,recovery]
class ScopeContractRefused(Exception):
    """Typed refusal raised when the reviewed scope contract cannot be read exactly."""

    # #region Tooling.DrEvidenceCheck.TypedScope.Scopecontractrefused.Init [C:3] [TYPE Function] [SEMANTICS dr,recovery]
    def __init__(self, code: str):
        self.code = code
        super().__init__(code)
    # #endregion Tooling.DrEvidenceCheck.TypedScope.Scopecontractrefused.Init
# #endregion Tooling.DrEvidenceCheck.TypedScope.Scopecontractrefused



# #region Tooling.DrEvidenceCheck.TypedScope.EntryEvidenceRefs [C:3] [TYPE Function] [SEMANTICS dr,recovery]
def _entry_evidence_refs(entry) -> list[dict]:
    """Extract DraftRef evidence fields from one typed catalog entry revision."""
    refs = []
    native = getattr(entry.entry, "native_result", None)
    if native is not None:
        record = native.authority_record
        refs.append({"ref": record.capture_content_ref, "sha256": record.source_response_hash,
                     "source": "catalog_capture_content_ref"})
        descriptor = record.descriptor.native_request
        refs.append({"ref": descriptor.content_ref, "sha256": descriptor.sha256,
                     "source": "catalog_descriptor_content_ref"})
    expected_image = getattr(entry.entry, "expected_image_content_ref", None)
    if expected_image:
        refs.append({"ref": expected_image, "sha256": expected_image.rsplit(":", 1)[-1],
                     "source": "catalog_expected_image_content_ref"})
    return refs
# #endregion Tooling.DrEvidenceCheck.TypedScope.EntryEvidenceRefs


# #region Tooling.DrEvidenceCheck.TypedScope.AuthorityEvidenceRefs [C:3] [TYPE Function] [SEMANTICS dr,recovery]
def _authority_evidence_refs(authority) -> list[dict]:
    """Extract protected evidence from the typed entry authority, including nested security evidence."""
    refs = []
    protected = getattr(authority, "protected_evidence_ref", None)
    if protected is not None:
        # The authoring envelope has no separate digest column: the DraftRef itself carries the digest.
        refs.append({"ref": protected, "sha256": None,
                     "source": "entry_authority_protected_evidence"})
    security = getattr(authority, "security_authority", None)
    if security is not None:
        refs.append({"ref": security.protected_evidence_ref,
                     "sha256": security.protected_evidence_sha256,
                     "source": "entry_authority_security_evidence"})
    return refs
# #endregion Tooling.DrEvidenceCheck.TypedScope.AuthorityEvidenceRefs


# #region Tooling.DrEvidenceCheck.TypedScope.MergeScope [C:3] [TYPE Function] [SEMANTICS dr,recovery]
def _merge_scope(rows: list[dict]) -> list[dict]:
    """One row per ref; conflicting digests for one ref are a typed refusal, never a silent pick."""
    merged: dict[str, dict] = {}
    for row in rows:
        ref, digest = row["ref"], row.get("sha256")
        existing = merged.get(ref)
        if existing is None:
            merged[ref] = row
            continue
        existing_digest = existing.get("sha256")
        if existing_digest is None:
            existing["sha256"] = digest
        elif digest is not None and existing_digest != digest:
            raise ScopeContractRefused("EVIDENCE_SCOPE_CONFLICT")
    return list(merged.values())
# #endregion Tooling.DrEvidenceCheck.TypedScope.MergeScope


# #region Tooling.DrEvidenceCheck.TypedScope.DocumentScopeRows [C:3] [TYPE Function] [SEMANTICS dr,recovery]
def _document_scope_rows(payload, CatalogRevisionDocument, authority_kinds) -> list[dict]:
    """Validate one stored document through the typed contract, then extract its evidence fields."""
    try:
        document = CatalogRevisionDocument.model_validate_json(
            json.dumps(payload), context={"entry_authority_kinds": authority_kinds})
    except Exception:  # noqa: BLE001 — the typed contract must validate, no fallback
        raise ScopeContractRefused("EVIDENCE_CATALOG_PAYLOAD_INVALID") from None
    rows: list[dict] = []
    for entry in document.entry_revisions + document.observatory_entries:
        rows.extend(_entry_evidence_refs(entry))
    return rows
# #endregion Tooling.DrEvidenceCheck.TypedScope.DocumentScopeRows


# #region Tooling.DrEvidenceCheck.TypedScope.AuthorityScopeRows [C:3] [TYPE Function] [SEMANTICS dr,recovery]
def _authority_scope_rows(payload, validate_authority) -> list[dict]:
    """Validate one stored authority envelope through the typed contract, then extract evidence."""
    try:
        authority = validate_authority(payload)
    except Exception:  # noqa: BLE001
        raise ScopeContractRefused("EVIDENCE_ENTRY_AUTHORITY_INVALID") from None
    return _authority_evidence_refs(authority)
# #endregion Tooling.DrEvidenceCheck.TypedScope.AuthorityScopeRows


# #region Tooling.DrEvidenceCheck.TypedScope.CatalogScopeRows [C:3] [TYPE Function] [SEMANTICS dr,recovery]
def _catalog_scope_rows(engine, CatalogRevisionDocument, validate_authority) -> list[dict]:
    from sqlalchemy import text

    rows: list[dict] = []
    with engine.connect() as connection:
        payloads = connection.execute(text(
            "SELECT payload FROM baseline_catalog_revisions")).scalars().all()
        authorities = connection.execute(text(
            "SELECT catalog_revision_id, baseline_revision_id, authority_kind, authority "
            "FROM baseline_catalog_entry_authorities")).mappings().all()
    kinds_by_revision: dict[str, dict[str, str]] = {}
    for item in authorities:
        kinds_by_revision.setdefault(item["catalog_revision_id"], {})[
            item["baseline_revision_id"]] = item["authority_kind"]
    for payload in payloads:
        revision_id = payload.get("catalog_revision_id") if isinstance(payload, dict) else None
        rows.extend(_document_scope_rows(payload, CatalogRevisionDocument,
                                         kinds_by_revision.get(revision_id, {})))
    for item in authorities:
        rows.extend(_authority_scope_rows(item["authority"], validate_authority))
    return rows
# #endregion Tooling.DrEvidenceCheck.TypedScope.CatalogScopeRows


# #region Tooling.DrEvidenceCheck.TypedScope.ScopeFromDatabase [C:3] [TYPE Function] [SEMANTICS dr,recovery]
def _scope_from_database(CatalogRevisionDocument, validate_authority) -> tuple[list[dict], int]:
    """Return in-scope rows and the out-of-scope draft count; read-only queries only."""
    from sqlalchemy import create_engine, text

    database_url = os.environ.get("DATABASE_URL", "")
    if not database_url:
        raise ScopeContractRefused("DR_TARGET_UNRESOLVED")
    schema = (os.environ.get("DB_SCHEMA") or "public").strip() or "public"
    connect_args = {} if schema == "public" else {"options": f"-csearch_path={schema}"}
    engine = create_engine(database_url, connect_args=connect_args)
    try:
        rows = _catalog_scope_rows(engine, CatalogRevisionDocument, validate_authority)
        with engine.connect() as connection:
            scenario_rows = connection.execute(text(
                "SELECT a.content_ref AS ref, a.sha256 AS sha256 FROM scenario_artifacts a "
                "LEFT JOIN scenario_retention_deletions d "
                "  ON d.target_type = 'artifact' AND d.target_id = a.id "
                " AND d.state IN ('deleted', 'tombstoned') "
                "WHERE a.is_active AND d.id IS NULL")).mappings().all()
            known = {row["ref"] for row in rows}
            for row in scenario_rows:
                rows.append({"ref": row["ref"], "sha256": row["sha256"],
                             "source": "active_scenario_artifact"})
                known.add(row["ref"])
            out_of_scope = 0
            for row in connection.execute(text(
                    "SELECT content_ref AS ref, sha256 AS sha256 FROM draft_artifacts")).mappings().all():
                if row["ref"] in known:
                    rows.append({"ref": row["ref"], "sha256": row["sha256"],
                                 "source": "referenced_draft_artifact"})
                else:
                    out_of_scope += 1
    finally:
        engine.dispose()
    return _merge_scope(rows), out_of_scope
# #endregion Tooling.DrEvidenceCheck.TypedScope.ScopeFromDatabase
# #endregion Tooling.DrEvidenceCheck.TypedScope