187 lines
9.0 KiB
Python
187 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
# #region Tooling.DrEvidenceCheck [C:4] [TYPE Module] [SEMANTICS dr,evidence,verification,offline,digest,scope]
|
|
# @ingroup Tooling
|
|
# @BRIEF Prove, offline and read-only, that every in-scope opaque evidence ref resolves to contained bytes whose sha256 matches.
|
|
# @RELATION DEPENDS_ON -> [Tooling.DrEvidenceCheck.TypedScope]
|
|
# @RELATION DEPENDS_ON -> [Tooling.DrEvidenceCheck.Containment]
|
|
# @PRE DATABASE_URL names the exact restored database and STORAGE_ROOT_PATH the archived evidence tree.
|
|
# @POST Nothing is written; the summary reports verified/out-of-scope counts and typed refusal codes without raw refs.
|
|
# @INVARIANT Scope comes from the typed catalog contract and typed columns, never from a prefix scan or a fallback.
|
|
# @INVARIANT Traversal uses pinned directory descriptors (dir_fd + O_NOFOLLOW per component), so replacing a
|
|
# verified parent directory with a symlink cannot redirect the open to external bytes.
|
|
# @INVARIANT A handle ref is accepted only with the explicit handle: prefix (HandleBlobStore.retrieve).
|
|
# @RATIONALE A missing or escaping byte must fail the DR check loudly; counters without digests cannot prove evidence.
|
|
# @REJECTED A second draft-ref parser is rejected: DraftStorage.parse_content_ref stays the single authority.
|
|
# @REJECTED Keyword-scanning payload text was rejected: it silently drops malformed refs in typed fields.
|
|
# @REJECTED Printing raw refs in diagnostics was rejected: refs are untrusted strings and may carry secrets.
|
|
# @REJECTED Treating an empty scope as success is rejected: there is nothing to verify, so it is not a PASS.
|
|
"""Offline DR evidence verification: typed scope -> contained storage paths -> sha256, fail-closed.
|
|
|
|
Scope comes from the reviewed typed contract:
|
|
* catalog revision payload DraftRef fields (capture_content_ref, descriptor native_request
|
|
content_ref, visual expected_image_content_ref) -> validated by CatalogRevisionDocument;
|
|
* entry-authority protected evidence, including the nested security authority evidence;
|
|
* scenario_artifacts rows that are active and have no terminal retention receipt;
|
|
* draft_artifacts rows whose content_ref is referenced by any of the above.
|
|
|
|
Exit codes:
|
|
0 every in-scope ref resolved, contained, exists and matches (checked > 0)
|
|
1 at least one typed refusal (see EVIDENCE_* codes)
|
|
3 no in-scope refs at all (nothing verified, never a PASS)
|
|
4 the production draft-ref parser could not be imported
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os as os
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
from dr_evidence_scope import (
|
|
ScopeContractRefused as ScopeContractRefused, _entry_evidence_refs as _entry_evidence_refs, _authority_evidence_refs as _authority_evidence_refs, _merge_scope as _merge_scope,
|
|
_document_scope_rows as _document_scope_rows, _authority_scope_rows as _authority_scope_rows, _catalog_scope_rows as _catalog_scope_rows, _scope_from_database as _scope_from_database,
|
|
)
|
|
from dr_evidence_io import (
|
|
_resolve as _resolve, _open_root as _open_root, _classify_component as _classify_component, _open_contained as _open_contained, _sha256_descriptor as _sha256_descriptor, _archive_names as _archive_names,
|
|
)
|
|
|
|
EXIT_OK = 0
|
|
EXIT_FAILURES = 1
|
|
EXIT_EMPTY_SCOPE = 3
|
|
EXIT_APP_IMPORT_FAILED = 4
|
|
|
|
REFERENCE_PREFIXES = ("draft:", "handle:")
|
|
TERMINAL_RETENTION_STATES = ("deleted", "tombstoned")
|
|
|
|
|
|
# #region Tooling.DrEvidenceCheck.App [C:2] [TYPE Function]
|
|
def _app_modules():
|
|
"""Import the production owners; a missing owner is a typed refusal, never a fallback parser."""
|
|
repo_root = Path(__file__).resolve().parents[1]
|
|
backend_root = repo_root / "backend"
|
|
if str(backend_root) not in sys.path:
|
|
sys.path.insert(0, str(backend_root))
|
|
try:
|
|
from src.schemas.dashboard_testing.catalog_revision import (
|
|
CatalogRevisionDocument, validate_catalog_entry_authority)
|
|
from src.services.agent_runs.artifacts import DraftStorage
|
|
except Exception as exc: # noqa: BLE001 — any import failure must stop the check
|
|
print(json.dumps({"code": "EVIDENCE_APP_IMPORT_FAILED", "reason": type(exc).__name__}))
|
|
raise SystemExit(EXIT_APP_IMPORT_FAILED) from None
|
|
return CatalogRevisionDocument, validate_catalog_entry_authority, DraftStorage.parse_content_ref
|
|
# #endregion Tooling.DrEvidenceCheck.App
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# #region Tooling.DrEvidenceCheck.Scheme [C:3] [TYPE Function] [SEMANTICS dr,recovery]
|
|
def _scheme(ref: str) -> str:
|
|
if ref.startswith("draft:"):
|
|
return "draft"
|
|
if ref.startswith("handle:"):
|
|
return "handle"
|
|
return "unsupported"
|
|
# #endregion Tooling.DrEvidenceCheck.Scheme
|
|
|
|
|
|
# #region Tooling.DrEvidenceCheck.Failure [C:3] [TYPE Function] [SEMANTICS dr,recovery]
|
|
def _failure(code: str, row: dict) -> dict:
|
|
"""Diagnostics carry only controlled identifiers and a ref digest; never the raw ref text."""
|
|
ref = row["ref"] if isinstance(row.get("ref"), str) else ""
|
|
return {"code": code, "source": row.get("source", "unknown"),
|
|
"scheme": _scheme(ref),
|
|
"ref_digest": hashlib.sha256(ref.encode("utf-8", "surrogatepass")).hexdigest()[:16]}
|
|
# #endregion Tooling.DrEvidenceCheck.Failure
|
|
|
|
|
|
# #region Tooling.DrEvidenceCheck.Verify [C:4] [TYPE Function]
|
|
def _verify(root, parse_draft, archive_names: set[str] | None,
|
|
rows: list[dict]) -> list[dict]:
|
|
root_real = _storage_root(str(root))
|
|
root_fd = _open_root(root_real)
|
|
failures: list[dict] = []
|
|
try:
|
|
for row in rows:
|
|
ref, row_sha = row.get("ref"), row.get("sha256")
|
|
if not isinstance(ref, str):
|
|
failures.append(_failure("EVIDENCE_REF_MALFORMED", row))
|
|
continue
|
|
try:
|
|
relative_parts, digest = _resolve(parse_draft, ref)
|
|
except (ValueError, TypeError):
|
|
failures.append(_failure("EVIDENCE_REF_MALFORMED", row))
|
|
continue
|
|
if row_sha is not None and row_sha != digest:
|
|
failures.append(_failure("EVIDENCE_ROW_DIGEST_MISMATCH", row))
|
|
relative = "/".join(relative_parts)
|
|
if archive_names is not None and relative not in archive_names:
|
|
failures.append(_failure("EVIDENCE_ARCHIVE_MISSING", row))
|
|
descriptor, containment = _open_contained(root_fd, relative_parts)
|
|
if descriptor is None:
|
|
failures.append(_failure(containment, row))
|
|
continue
|
|
if _sha256_descriptor(descriptor) != digest:
|
|
failures.append(_failure("EVIDENCE_DIGEST_MISMATCH", row))
|
|
finally:
|
|
os.close(root_fd)
|
|
return failures
|
|
# #endregion Tooling.DrEvidenceCheck.Verify
|
|
|
|
|
|
# #region Tooling.DrEvidenceCheck.StorageRoot [C:3] [TYPE Function] [SEMANTICS dr,recovery]
|
|
def _storage_root(value: str) -> Path:
|
|
candidate = Path(value)
|
|
if not candidate.is_absolute() or not candidate.is_dir():
|
|
raise ScopeContractRefused("EVIDENCE_STORAGE_ROOT_INVALID")
|
|
return candidate.resolve(strict=True)
|
|
# #endregion Tooling.DrEvidenceCheck.StorageRoot
|
|
|
|
|
|
# #region Tooling.DrEvidenceCheck.Main [C:3] [TYPE Function]
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Read-only DR evidence verifier (no DB or storage writes).")
|
|
parser.add_argument("--storage-root", default=os.environ.get("STORAGE_ROOT_PATH", "/app/storage"))
|
|
parser.add_argument("--archive-list", default=None,
|
|
help="output of 'tar --list --gzip --file <storage archive>'")
|
|
parser.add_argument("--refs-from", default=None,
|
|
help="offline JSON list of {ref, sha256, source}; skips the database")
|
|
args = parser.parse_args(argv)
|
|
CatalogRevisionDocument, validate_authority, parse_draft = _app_modules()
|
|
try:
|
|
root_real = _storage_root(args.storage_root)
|
|
archive_names = _archive_names(args.archive_list)
|
|
if args.refs_from:
|
|
rows, out_of_scope = json.loads(Path(args.refs_from).read_text(encoding="utf-8")), 0
|
|
if not isinstance(rows, list):
|
|
raise ScopeContractRefused("EVIDENCE_REFS_FIXTURE_INVALID")
|
|
else:
|
|
rows, out_of_scope = _scope_from_database(CatalogRevisionDocument, validate_authority)
|
|
except ScopeContractRefused as refusal:
|
|
print(json.dumps({"code": refusal.code, "failures": [], "codes": [refusal.code]}))
|
|
return EXIT_FAILURES
|
|
failures = _verify(root_real, parse_draft, archive_names, rows)
|
|
codes = sorted({item["code"] for item in failures})
|
|
if not rows and not failures:
|
|
codes = ["EVIDENCE_SCOPE_EMPTY"]
|
|
summary = {"checked": len(rows), "out_of_scope": out_of_scope,
|
|
"archive_list_checked": archive_names is not None,
|
|
"failures": failures, "codes": codes}
|
|
print(json.dumps(summary, ensure_ascii=False))
|
|
if failures:
|
|
return EXIT_FAILURES
|
|
if not rows:
|
|
return EXIT_EMPTY_SCOPE
|
|
return EXIT_OK
|
|
# #endregion Tooling.DrEvidenceCheck.Main
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
# #endregion Tooling.DrEvidenceCheck
|