← Files ClaraARCHIVED FILE
scripts/validate_real_data_pilot_intake.py
57.5 KB · Oct 2, 2026 · 00:29 UTC
#!/usr/bin/env python3
"""Validate the declared authorization boundary for a real-data pilot.
The checks in this module are deterministic because exact source binding,
declared purpose, dates, and fixed privacy/publication boundaries are
mechanically verifiable. They do not establish that an authorizer has legal
authority, that data is sufficiently anonymized, or that accounting semantics
are correct. Those remain reviewed human judgements.
"""
from __future__ import annotations
import argparse
import json
import logging
import os
import re
import shutil
import stat
# The only subprocess is a fixed, shell-free git check-ignore invocation.
import subprocess # nosec B404
from collections.abc import Mapping, Sequence
from contextlib import contextmanager
from datetime import date
from pathlib import Path
from typing import Any, Iterator
from preparation_contract_kernel import (
ContractValidationError,
canonical_json_sha256,
file_sha256,
file_snapshot_beneath,
strict_json_snapshot_beneath,
)
__all__ = [
"INTAKE_RECEIPT_SCHEMA",
"INTAKE_RECEIPT_SCHEMA_V2",
"INTAKE_SCHEMA",
"INTAKE_SCHEMA_V2",
"PILOT_ID_PATTERN",
"pinned_pilot_receipt_output",
"resolve_pilot_storage_roots",
"validate_pilot_receipt_output_path",
"validate_pilot_local_run_path",
"validate_real_data_pilot_intake_receipt",
"validate_real_data_pilot_intake_receipt_v2",
"validate_real_data_pilot_intake",
"validate_real_data_pilot_intake_v2",
"main",
]
LOGGER = logging.getLogger(__name__)
INTAKE_SCHEMA = "clara.real_data_pilot_intake.v1"
INTAKE_RECEIPT_SCHEMA = "clara.real_data_pilot_intake_receipt.v1"
INTAKE_SCHEMA_V2 = "clara.real_data_pilot_intake.v2"
INTAKE_RECEIPT_SCHEMA_V2 = "clara.real_data_pilot_intake_receipt.v2"
VALIDATOR_VERSION = "1.0.0"
VALIDATOR_ID = "real_data_pilot_intake_validator.v1"
VALIDATOR_VERSION_V2 = "2.0.0"
VALIDATOR_ID_V2 = "real_data_pilot_intake_validator.v2"
PURPOSE = "local_due_diligence_preparation_evaluation"
SOURCE_KIND = "commercial_trial_balance"
SOURCE_KINDS_V2 = frozenset(
{
"commercial_general_ledger",
"commercial_trial_balance",
}
)
DATA_CLASSIFICATIONS = frozenset({"anonymized_real", "consented_real"})
AUTHORIZATION_BASES = frozenset(
{
"explicit_authorized_user_instruction",
"public_owner_license",
"written_data_owner_permission",
}
)
PERMITTED_ACTIONS = (
"codex_model_processing",
"local_deterministic_processing",
)
PROHIBITED_ACTIONS = (
"commit_raw_or_row_level_data",
"package_raw_or_row_level_data",
"publish_raw_or_row_level_data",
)
REQUIRED_SEMANTIC_REVIEWS = (
"account_mapping",
"control_equivalence_and_tolerance",
"currency_unit_and_fx",
"dataset_identity_and_grain",
"period_calendar_and_value_basis",
"scope_entity_and_eliminations",
"sign_convention",
)
DOES_NOT_ESTABLISH = (
"accounting_semantic_correctness",
"anonymization_sufficiency",
"authorizer_identity_or_authority",
"commercial_origin_or_authenticity",
"current_authorization_at_later_execution",
"downstream_compatibility",
"future_storage_or_copy_behavior",
"legal_permission",
"publication_authorization",
"receipt_authenticity_or_signer_identity",
"retention_or_deletion",
"report_readiness",
)
DOES_NOT_ESTABLISH_V2 = (
"accounting_semantic_correctness",
"anonymization_sufficiency",
"authorizer_identity_or_authority",
"commercial_origin_or_authenticity",
"current_authorization_at_later_execution",
"declared_data_kind_correctness",
"downstream_compatibility",
"future_storage_or_copy_behavior",
"legal_permission",
"publication_authorization",
"receipt_authenticity_or_signer_identity",
"retention_or_deletion",
"report_readiness",
)
PILOT_ID_PATTERN = re.compile(r"^pilot-[0-9a-f]{16}$")
SOURCE_ID_PATTERN = re.compile(r"^source-[0-9a-f]{16}$")
SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
MEDIA_TYPES = frozenset(
{
"application/vnd.ms-excel",
"application/vnd.oasis.opendocument.spreadsheet",
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
"text/csv",
}
)
STORAGE_RELATIONS = frozenset(
{
"inside_repository_git_ignored",
"outside_repository_within_run_root",
}
)
INTAKE_RECEIPT_FIELDS = frozenset(
{
"schema_version",
"pilot_id",
"validation_date",
"intake_contract_sha256",
"source_receipt",
"authorization_receipt",
"privacy_boundary",
"deidentification_review",
"semantic_review",
"eligibility",
"validator",
}
)
CLARA_ROOT = Path(__file__).resolve().parents[1]
INTAKE_VALIDATOR_DEPENDENCY_PATHS = {
"intake_receipt_schema": (
CLARA_ROOT / "contracts" / "real_data_pilot_intake_receipt.v1.schema.json"
),
"intake_schema": (
CLARA_ROOT / "contracts" / "real_data_pilot_intake.v1.schema.json"
),
"preparation_contract_kernel": (
Path(__file__).resolve().with_name("preparation_contract_kernel.py")
),
}
INTAKE_VALIDATOR_DEPENDENCY_PATHS_V2 = {
"intake_receipt_schema": (
CLARA_ROOT / "contracts" / "real_data_pilot_intake_receipt.v2.schema.json"
),
"intake_schema": (
CLARA_ROOT / "contracts" / "real_data_pilot_intake.v2.schema.json"
),
"preparation_contract_kernel": (
Path(__file__).resolve().with_name("preparation_contract_kernel.py")
),
}
def _mapping(value: Any, *, label: str) -> Mapping[str, Any]:
if not isinstance(value, Mapping):
raise ContractValidationError(f"{label} must be an object")
return value
def _text(value: Any, *, label: str) -> str:
if not isinstance(value, str) or not value or value != value.strip():
raise ContractValidationError(
f"{label} must be canonical non-empty text without edge whitespace"
)
return value
def _opaque_identifier(
value: Any,
*,
pattern: re.Pattern[str],
label: str,
) -> str:
result = _text(value, label=label)
if pattern.fullmatch(result) is None:
raise ContractValidationError(f"{label} must be an opaque digest-shaped ID")
return result
def _iso_date(value: Any, *, label: str) -> date:
result = _text(value, label=label)
if re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", result) is None:
raise ContractValidationError(f"{label} must be an ISO date")
try:
return date.fromisoformat(result)
except ValueError as exc:
raise ContractValidationError(f"{label} must be an ISO date") from exc
def _optional_iso_date(value: Any, *, label: str) -> date | None:
if value is None:
return None
return _iso_date(value, label=label)
def _exact_fields(
value: Mapping[str, Any],
*,
required: frozenset[str],
label: str,
) -> None:
missing = sorted(required - set(value))
unexpected = sorted(set(value) - required)
if missing:
raise ContractValidationError(f"{label} is missing fields: {missing}")
if unexpected:
raise ContractValidationError(
f"{label} contains unexpected fields: {unexpected}"
)
def _exact_text_list(
value: Any,
*,
expected: tuple[str, ...],
label: str,
) -> list[str]:
if not isinstance(value, list) or any(not isinstance(item, str) for item in value):
raise ContractValidationError(f"{label} must be a list of text values")
if value != list(expected):
raise ContractValidationError(
f"{label} must equal the registered ordered values"
)
return list(value)
def _require_constant(value: Any, *, expected: Any, label: str) -> None:
if type(value) is not type(expected) or value != expected:
raise ContractValidationError(f"{label} must equal {expected!r}")
def _current_dependency_sha256(
dependency_paths: Mapping[str, Path] = INTAKE_VALIDATOR_DEPENDENCY_PATHS,
) -> dict[str, str]:
"""Return hashes for every file that can change intake validation behavior."""
return {
dependency_id: file_sha256(path)
for dependency_id, path in sorted(dependency_paths.items())
}
def _validate_dependency_sha256(
value: Any,
*,
label: str,
dependency_paths: Mapping[str, Path] = INTAKE_VALIDATOR_DEPENDENCY_PATHS,
) -> dict[str, str]:
dependencies = _mapping(value, label=label)
_exact_fields(
dependencies,
required=frozenset(dependency_paths),
label=label,
)
expected = _current_dependency_sha256(dependency_paths)
validated: dict[str, str] = {}
for dependency_id, expected_digest in expected.items():
digest = _receipt_sha256(
dependencies[dependency_id],
label=f"{label}.{dependency_id}",
)
if digest != expected_digest:
raise ContractValidationError(f"{label}.{dependency_id} is not current")
validated[dependency_id] = digest
return validated
def _current_date() -> date:
"""Return the CLI validation date through a testable time seam."""
return date.today()
def _enclosing_git_worktree(path: Path) -> Path | None:
resolved_path = Path(path).resolve()
for candidate in (resolved_path, *resolved_path.parents):
if (candidate / ".git").exists():
return candidate
return None
def resolve_pilot_storage_roots(
*,
local_run_root: Path,
repository_root: Path,
) -> tuple[Path, Path]:
run_root = Path(local_run_root).resolve()
repo_root = Path(repository_root).resolve()
if not run_root.is_dir():
raise ContractValidationError("local_run_root must be an existing directory")
if not repo_root.is_dir() or not (repo_root / ".git").exists():
raise ContractValidationError(
"repository_root must identify an existing Git worktree"
)
enclosing_worktree = _enclosing_git_worktree(run_root)
if enclosing_worktree is not None and enclosing_worktree != repo_root:
raise ContractValidationError(
"repository_root must match the Git worktree enclosing local_run_root"
)
return run_root, repo_root
def _require_git_ignored(path: Path, *, repository_root: Path, label: str) -> None:
git_path = shutil.which("git")
if git_path is None:
raise ContractValidationError(
f"{label} is inside the repository but Git is unavailable"
)
relative_path = path.relative_to(repository_root)
# The executable is resolved locally and argv is fixed; no shell is used.
result = subprocess.run( # nosec B603
[
git_path,
"-C",
str(repository_root),
"check-ignore",
"--quiet",
"--",
str(relative_path),
],
check=False,
capture_output=True,
)
if result.returncode == 1:
raise ContractValidationError(
f"{label} is inside the repository but is not Git-ignored"
)
if result.returncode != 0:
raise ContractValidationError(
f"Git could not verify the ignored status of {label}"
)
def validate_pilot_local_run_path(
path: Path,
*,
local_run_root: Path,
repository_root: Path,
label: str,
) -> tuple[Path, str]:
resolved_path = Path(path).resolve()
if not resolved_path.is_relative_to(local_run_root):
raise ContractValidationError(
f"{label} must resolve inside the declared local run root"
)
if resolved_path.is_file() and resolved_path.stat().st_nlink != 1:
raise ContractValidationError(f"{label} must not be a hard-linked file")
enclosing_worktree = _enclosing_git_worktree(resolved_path)
if enclosing_worktree is not None and enclosing_worktree != repository_root:
raise ContractValidationError(
f"{label} must not be inside an undeclared Git worktree"
)
if enclosing_worktree == repository_root:
_require_git_ignored(
resolved_path,
repository_root=repository_root,
label=label,
)
return resolved_path, "inside_repository_git_ignored"
return resolved_path, "outside_repository_within_run_root"
def validate_pilot_receipt_output_path(
path: Path,
*,
declared_local_run_root: Path,
local_run_root: Path,
repository_root: Path,
label: str,
) -> tuple[Path, str]:
"""Resolve one output path while rejecting symlinks below its declared root."""
lexical_root = Path(declared_local_run_root).absolute()
lexical_path = Path(path).absolute()
if lexical_root.is_symlink():
raise ContractValidationError("local_run_root must not identify a symlink")
try:
relative_parts = lexical_path.relative_to(lexical_root).parts
except ValueError:
relative_parts = ()
cursor = lexical_root
for part in relative_parts:
cursor /= part
if cursor.is_symlink():
raise ContractValidationError(
f"{label} must not use a symlink below local_run_root"
)
if lexical_path.is_symlink():
raise ContractValidationError(f"{label} must not identify a symlink")
return validate_pilot_local_run_path(
lexical_path,
local_run_root=local_run_root,
repository_root=repository_root,
label=label,
)
def _descriptor_identity(value: os.stat_result) -> tuple[int, int, int]:
"""Return the stable identity needed to detect directory-entry swaps."""
return value.st_dev, value.st_ino, stat.S_IFMT(value.st_mode)
def _file_snapshot_identity(
value: os.stat_result,
) -> tuple[int, int, int, int, int, int, int]:
"""Return file identity plus mutation-sensitive metadata."""
return (
value.st_dev,
value.st_ino,
stat.S_IFMT(value.st_mode),
value.st_nlink,
value.st_size,
value.st_mtime_ns,
value.st_ctime_ns,
)
def _read_descriptor_bytes(descriptor: int) -> bytes:
chunks: list[bytes] = []
while True:
chunk = os.read(descriptor, 1024 * 1024)
if not chunk:
return b"".join(chunks)
chunks.append(chunk)
class _PinnedPilotReceiptOutput:
"""Write one receipt through a no-follow directory descriptor chain."""
def __init__(
self,
*,
output_path: Path,
root_path: Path,
relative_parts: tuple[str, ...],
directory_descriptors: tuple[int, ...],
) -> None:
self.output_path = output_path
self.root_path = root_path
self.relative_parts = relative_parts
self.directory_descriptors = directory_descriptors
@property
def _parent_descriptor(self) -> int:
return self.directory_descriptors[-1]
@property
def _filename(self) -> str:
return self.relative_parts[-1]
def _verify_directory_chain(self) -> None:
"""Fail if any lexical directory entry no longer names its pinned inode."""
try:
root_entry = self.root_path.lstat()
except OSError as exc:
raise ContractValidationError(
"local_run_root changed during receipt write"
) from exc
if _descriptor_identity(root_entry) != _descriptor_identity(
os.fstat(self.directory_descriptors[0])
):
raise ContractValidationError("local_run_root changed during receipt write")
for position, part in enumerate(self.relative_parts[:-1]):
try:
current_entry = os.stat(
part,
dir_fd=self.directory_descriptors[position],
follow_symlinks=False,
)
except OSError as exc:
raise ContractValidationError(
"output directory path changed during receipt write"
) from exc
pinned_entry = os.fstat(self.directory_descriptors[position + 1])
if _descriptor_identity(current_entry) != _descriptor_identity(
pinned_entry
):
raise ContractValidationError(
"output directory path changed during receipt write"
)
def require_absent(self) -> None:
"""Require a fresh output leaf; never replace or delete existing data."""
try:
os.stat(
self._filename,
dir_fd=self._parent_descriptor,
follow_symlinks=False,
)
except FileNotFoundError:
return
except OSError as exc:
raise ContractValidationError(
"output path could not be checked before receipt write"
) from exc
raise ContractValidationError(
"output path must be absent; choose a fresh receipt path"
)
def write_json(self, value: Mapping[str, Any]) -> None:
"""Write exact JSON through one new no-follow output descriptor.
A failure never deletes or replaces the output name. A partial or
concurrently changed leaf is left in place for explicit review.
"""
# The structured-value check is deterministic because JSON safety and
# exact output bytes are mechanically verifiable contract properties.
canonical_json_sha256(value)
output_bytes = (
json.dumps(
value,
ensure_ascii=False,
indent=2,
sort_keys=True,
allow_nan=False,
)
+ "\n"
).encode("utf-8")
self._verify_directory_chain()
self.require_absent()
output_flags = os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW
if hasattr(os, "O_CLOEXEC"):
output_flags |= os.O_CLOEXEC
try:
output_descriptor = os.open(
self._filename,
output_flags,
0o600,
dir_fd=self._parent_descriptor,
)
except FileExistsError as exc:
raise ContractValidationError(
"output path became occupied before receipt write"
) from exc
except OSError as exc:
raise ContractValidationError(
"receipt output could not be safely created"
) from exc
try:
remaining = memoryview(output_bytes)
while remaining:
written = os.write(output_descriptor, remaining)
if written <= 0:
raise ContractValidationError(
"receipt output could not be written completely"
)
remaining = remaining[written:]
os.fsync(output_descriptor)
descriptor_before = os.fstat(output_descriptor)
if (
not stat.S_ISREG(descriptor_before.st_mode)
or descriptor_before.st_nlink != 1
or descriptor_before.st_size != len(output_bytes)
):
raise ContractValidationError(
"receipt output failed its integrity check"
)
os.lseek(output_descriptor, 0, os.SEEK_SET)
observed_bytes = _read_descriptor_bytes(output_descriptor)
descriptor_after = os.fstat(output_descriptor)
if observed_bytes != output_bytes or _file_snapshot_identity(
descriptor_before
) != _file_snapshot_identity(descriptor_after):
raise ContractValidationError(
"receipt output bytes changed during the write"
)
self._verify_directory_chain()
try:
current_output = os.stat(
self._filename,
follow_symlinks=False,
dir_fd=self._parent_descriptor,
)
except OSError as exc:
raise ContractValidationError(
"receipt output path changed during the write"
) from exc
if (
_descriptor_identity(current_output)
!= _descriptor_identity(descriptor_after)
or current_output.st_nlink != 1
or current_output.st_size != len(output_bytes)
):
raise ContractValidationError(
"published receipt failed its integrity check"
)
os.fsync(self._parent_descriptor)
self._verify_directory_chain()
final_output = os.stat(
self._filename,
dir_fd=self._parent_descriptor,
follow_symlinks=False,
)
final_descriptor_before = os.fstat(output_descriptor)
os.lseek(output_descriptor, 0, os.SEEK_SET)
final_bytes = _read_descriptor_bytes(output_descriptor)
final_descriptor_after = os.fstat(output_descriptor)
if (
final_bytes != output_bytes
or _file_snapshot_identity(final_output)
!= _file_snapshot_identity(final_descriptor_before)
or _file_snapshot_identity(final_descriptor_before)
!= _file_snapshot_identity(final_descriptor_after)
):
raise ContractValidationError(
"receipt output changed before write completion"
)
self._verify_directory_chain()
finally:
os.close(output_descriptor)
@contextmanager
def pinned_pilot_receipt_output(
output_path: Path,
*,
local_run_root: Path,
) -> Iterator[_PinnedPilotReceiptOutput]:
"""Pin a root-relative no-follow parent directory for one receipt write."""
root_path = Path(local_run_root).absolute()
lexical_output = Path(output_path).absolute()
try:
relative_path = lexical_output.relative_to(root_path)
except ValueError as exc:
raise ContractValidationError(
"output path must stay inside the declared local run root"
) from exc
relative_parts = relative_path.parts
if not relative_parts or any(part in {"", ".", ".."} for part in relative_parts):
raise ContractValidationError(
"output path must be one root-relative receipt file"
)
directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
if hasattr(os, "O_CLOEXEC"):
directory_flags |= os.O_CLOEXEC
descriptors: list[int] = []
try:
current_descriptor = os.open(root_path, directory_flags)
descriptors.append(current_descriptor)
for part in relative_parts[:-1]:
current_descriptor = os.open(
part,
directory_flags,
dir_fd=current_descriptor,
)
descriptors.append(current_descriptor)
pinned_output = _PinnedPilotReceiptOutput(
output_path=lexical_output,
root_path=root_path,
relative_parts=relative_parts,
directory_descriptors=tuple(descriptors),
)
pinned_output._verify_directory_chain()
yield pinned_output
except OSError as exc:
raise ContractValidationError(
"output path parent could not be pinned below local_run_root"
) from exc
finally:
for descriptor in reversed(descriptors):
os.close(descriptor)
def _validate_source_kind(
value: Any,
*,
allowed: frozenset[str],
label: str,
) -> str:
source_kind = _text(value, label=label)
if source_kind not in allowed:
raise ContractValidationError(f"{label} is not registered")
return source_kind
def _validate_source_declaration(
value: Any,
*,
source_kinds: frozenset[str],
) -> dict[str, Any]:
source = _mapping(value, label="source")
_exact_fields(
source,
required=frozenset(
{
"source_id",
"data_kind",
"data_classification",
"media_type",
"byte_count",
"sha256",
}
),
label="source",
)
source_id = _opaque_identifier(
source["source_id"],
pattern=SOURCE_ID_PATTERN,
label="source.source_id",
)
source_kind = _validate_source_kind(
source["data_kind"],
allowed=source_kinds,
label="source.data_kind",
)
data_classification = _text(
source["data_classification"],
label="source.data_classification",
)
if data_classification not in DATA_CLASSIFICATIONS:
raise ContractValidationError(
"source.data_classification must be anonymized_real or consented_real"
)
media_type = _text(source["media_type"], label="source.media_type")
if media_type not in MEDIA_TYPES:
raise ContractValidationError("source.media_type is not registered")
byte_count = source["byte_count"]
if (
isinstance(byte_count, bool)
or not isinstance(byte_count, int)
or byte_count <= 0
):
raise ContractValidationError("source.byte_count must be a positive integer")
digest = _text(source["sha256"], label="source.sha256")
if SHA256_PATTERN.fullmatch(digest) is None:
raise ContractValidationError(
"source.sha256 must be a lowercase SHA-256 digest"
)
return {
"source_id": source_id,
"data_kind": source_kind,
"data_classification": data_classification,
"media_type": media_type,
"byte_count": byte_count,
"sha256": digest,
}
def _bind_source_file(
source: Mapping[str, Any],
*,
source_path: Path,
local_run_root: Path,
) -> dict[str, Any]:
source_path = Path(source_path)
if not source_path.is_file():
raise ContractValidationError("source path must identify one existing file")
actual_byte_count, actual_digest = file_snapshot_beneath(
source_path,
root=local_run_root,
)
if source["byte_count"] != actual_byte_count:
raise ContractValidationError(
"source.byte_count does not match the exact supplied file"
)
if source["sha256"] != actual_digest:
raise ContractValidationError("source.sha256 does not match the supplied file")
return {
"source_id": source["source_id"],
"declared_data_kind": source["data_kind"],
"declared_data_classification": source["data_classification"],
"media_type": source["media_type"],
"byte_count": source["byte_count"],
"sha256": source["sha256"],
}
def _validate_authorization(
value: Any,
*,
source_sha256: str,
as_of: date,
) -> dict[str, Any]:
authorization = _mapping(value, label="authorization")
_exact_fields(
authorization,
required=frozenset(
{
"status",
"basis",
"authority_assertion",
"evidence_reference",
"authorizing_role",
"authorized_on",
"valid_from",
"valid_until",
"purpose",
"authorized_source_sha256",
"permitted_actions",
"prohibited_actions",
"terms_summary",
}
),
label="authorization",
)
_require_constant(
authorization["status"],
expected="reviewed",
label="authorization.status",
)
basis = _text(authorization["basis"], label="authorization.basis")
if basis not in AUTHORIZATION_BASES:
raise ContractValidationError("authorization.basis is not registered")
_require_constant(
authorization["authority_assertion"],
expected="authorizer_has_right_to_permit_this_use",
label="authorization.authority_assertion",
)
_text(
authorization["evidence_reference"],
label="authorization.evidence_reference",
)
_text(
authorization["authorizing_role"],
label="authorization.authorizing_role",
)
authorized_on = _iso_date(
authorization["authorized_on"],
label="authorization.authorized_on",
)
valid_from = _iso_date(
authorization["valid_from"],
label="authorization.valid_from",
)
valid_until = _optional_iso_date(
authorization["valid_until"],
label="authorization.valid_until",
)
if authorized_on > as_of:
raise ContractValidationError(
"authorization.authorized_on cannot follow the validation date"
)
if valid_from > as_of:
raise ContractValidationError("authorization is not yet effective")
if valid_until is not None and valid_until < valid_from:
raise ContractValidationError(
"authorization.valid_until cannot precede valid_from"
)
if valid_until is not None and valid_until < as_of:
raise ContractValidationError("authorization has expired")
_require_constant(
authorization["purpose"],
expected=PURPOSE,
label="authorization.purpose",
)
_require_constant(
authorization["authorized_source_sha256"],
expected=source_sha256,
label="authorization.authorized_source_sha256",
)
_exact_text_list(
authorization["permitted_actions"],
expected=PERMITTED_ACTIONS,
label="authorization.permitted_actions",
)
_exact_text_list(
authorization["prohibited_actions"],
expected=PROHIBITED_ACTIONS,
label="authorization.prohibited_actions",
)
_text(authorization["terms_summary"], label="authorization.terms_summary")
return {
"status": "declared_reviewed",
"basis": basis,
"authorized_on": authorized_on.isoformat(),
"valid_from": valid_from.isoformat(),
"valid_until": valid_until.isoformat() if valid_until is not None else None,
"purpose": PURPOSE,
"authorized_source_sha256": source_sha256,
"permitted_actions": list(PERMITTED_ACTIONS),
"prohibited_actions": list(PROHIBITED_ACTIONS),
"authorization_content_sha256": canonical_json_sha256(authorization),
"assurance": ("declaration_bound_to_exact_source_not_independently_verified"),
}
def _validate_privacy(value: Any) -> dict[str, Any]:
privacy = _mapping(value, label="privacy")
_exact_fields(
privacy,
required=frozenset(
{
"codex_context_acknowledged",
"automatic_anonymization_claimed",
"clara_external_recipient_added",
"raw_and_row_level_storage",
"repository_recording_policy",
}
),
label="privacy",
)
_require_constant(
privacy["codex_context_acknowledged"],
expected=True,
label="privacy.codex_context_acknowledged",
)
_require_constant(
privacy["automatic_anonymization_claimed"],
expected=False,
label="privacy.automatic_anonymization_claimed",
)
_require_constant(
privacy["clara_external_recipient_added"],
expected=False,
label="privacy.clara_external_recipient_added",
)
_require_constant(
privacy["raw_and_row_level_storage"],
expected="local_run_root_only",
label="privacy.raw_and_row_level_storage",
)
_require_constant(
privacy["repository_recording_policy"],
expected="sanitized_summary_and_receipts_only",
label="privacy.repository_recording_policy",
)
return {
"codex_context_acknowledged": True,
"automatic_anonymization_claimed": False,
"clara_external_recipient_added": False,
"declared_raw_and_row_level_storage": "local_run_root_only",
"repository_recording_policy": "sanitized_summary_and_receipts_only",
}
def _validate_deidentification_review(
value: Any,
*,
data_classification: str,
) -> dict[str, Any]:
review = _mapping(value, label="deidentification_review")
_exact_fields(
review,
required=frozenset(
{
"status",
"basis",
"reidentification_risk_review_status",
}
),
label="deidentification_review",
)
status = _text(review["status"], label="deidentification_review.status")
risk_status = _text(
review["reidentification_risk_review_status"],
label="deidentification_review.reidentification_risk_review_status",
)
basis = _text(review["basis"], label="deidentification_review.basis")
if data_classification == "anonymized_real":
if status != "reviewed" or risk_status != "reviewed":
raise ContractValidationError(
"anonymized_real requires reviewed de-identification and "
"re-identification-risk records"
)
elif (status, risk_status) not in {
("not_applicable", "not_applicable"),
("reviewed", "reviewed"),
}:
raise ContractValidationError(
"consented_real de-identification and re-identification-risk "
"review statuses must match"
)
return {
"status": status,
"reidentification_risk_review_status": risk_status,
"review_content_sha256": canonical_json_sha256(review),
"assurance": "review_presence_only_not_anonymization_certification",
"basis_recorded": bool(basis),
}
def _validate_semantic_review_plan(value: Any) -> dict[str, Any]:
plan = _mapping(value, label="semantic_review_plan")
_exact_fields(
plan,
required=frozenset(
{
"status",
"required_reviews",
"automatic_mapping_allowed",
"unresolved_blocking_issues_block_preparation",
}
),
label="semantic_review_plan",
)
_require_constant(
plan["status"],
expected="pending",
label="semantic_review_plan.status",
)
required_reviews = _exact_text_list(
plan["required_reviews"],
expected=REQUIRED_SEMANTIC_REVIEWS,
label="semantic_review_plan.required_reviews",
)
_require_constant(
plan["automatic_mapping_allowed"],
expected=False,
label="semantic_review_plan.automatic_mapping_allowed",
)
_require_constant(
plan["unresolved_blocking_issues_block_preparation"],
expected=True,
label="semantic_review_plan.unresolved_blocking_issues_block_preparation",
)
return {
"status": "not_assessed",
"required_review_count": len(required_reviews),
"automatic_mapping_allowed": False,
"unresolved_blocking_issues_block_preparation": True,
}
def _receipt_sha256(value: Any, *, label: str) -> str:
digest = _text(value, label=label)
if SHA256_PATTERN.fullmatch(digest) is None:
raise ContractValidationError(f"{label} must be a lowercase SHA-256 digest")
return digest
def _validate_real_data_pilot_intake_receipt(
value: Any,
*,
receipt_schema: str,
source_kinds: frozenset[str],
does_not_establish: tuple[str, ...],
validator_id: str,
validator_version: str,
dependency_paths: Mapping[str, Path],
) -> dict[str, Any]:
"""Validate every field in one versioned sanitized intake receipt."""
receipt = _mapping(value, label="intake receipt")
_exact_fields(
receipt,
required=INTAKE_RECEIPT_FIELDS,
label="intake receipt",
)
_require_constant(
receipt["schema_version"],
expected=receipt_schema,
label="intake receipt.schema_version",
)
_opaque_identifier(
receipt["pilot_id"],
pattern=PILOT_ID_PATTERN,
label="intake receipt.pilot_id",
)
validation_date = _iso_date(
receipt["validation_date"],
label="intake receipt.validation_date",
)
_receipt_sha256(
receipt["intake_contract_sha256"],
label="intake receipt.intake_contract_sha256",
)
source = _mapping(receipt["source_receipt"], label="intake receipt.source_receipt")
_exact_fields(
source,
required=frozenset(
{
"source_id",
"declared_data_kind",
"declared_data_classification",
"media_type",
"byte_count",
"sha256",
}
),
label="intake receipt.source_receipt",
)
_opaque_identifier(
source["source_id"],
pattern=SOURCE_ID_PATTERN,
label="intake receipt.source_receipt.source_id",
)
_validate_source_kind(
source["declared_data_kind"],
allowed=source_kinds,
label="intake receipt.source_receipt.declared_data_kind",
)
data_classification = _text(
source["declared_data_classification"],
label="intake receipt.source_receipt.declared_data_classification",
)
if data_classification not in DATA_CLASSIFICATIONS:
raise ContractValidationError(
"intake receipt source classification is not registered"
)
media_type = _text(
source["media_type"],
label="intake receipt.source_receipt.media_type",
)
if media_type not in MEDIA_TYPES:
raise ContractValidationError(
"intake receipt source media_type is not registered"
)
byte_count = source["byte_count"]
if (
isinstance(byte_count, bool)
or not isinstance(byte_count, int)
or byte_count <= 0
):
raise ContractValidationError(
"intake receipt source byte_count must be a positive integer"
)
_receipt_sha256(
source["sha256"],
label="intake receipt.source_receipt.sha256",
)
authorization = _mapping(
receipt["authorization_receipt"],
label="intake receipt.authorization_receipt",
)
_exact_fields(
authorization,
required=frozenset(
{
"status",
"basis",
"authorized_on",
"valid_from",
"valid_until",
"purpose",
"authorized_source_sha256",
"permitted_actions",
"prohibited_actions",
"authorization_content_sha256",
"assurance",
}
),
label="intake receipt.authorization_receipt",
)
_require_constant(
authorization["status"],
expected="declared_reviewed",
label="intake receipt.authorization_receipt.status",
)
basis = _text(
authorization["basis"],
label="intake receipt.authorization_receipt.basis",
)
if basis not in AUTHORIZATION_BASES:
raise ContractValidationError(
"intake receipt authorization basis is not registered"
)
authorized_on = _iso_date(
authorization["authorized_on"],
label="intake receipt.authorization_receipt.authorized_on",
)
valid_from = _iso_date(
authorization["valid_from"],
label="intake receipt.authorization_receipt.valid_from",
)
valid_until = _optional_iso_date(
authorization["valid_until"],
label="intake receipt.authorization_receipt.valid_until",
)
if authorized_on > validation_date or valid_from > validation_date:
raise ContractValidationError(
"intake receipt authorization was not current at validation"
)
if valid_until is not None and (
valid_until < valid_from or valid_until < validation_date
):
raise ContractValidationError(
"intake receipt authorization was expired at validation"
)
_require_constant(
authorization["purpose"],
expected=PURPOSE,
label="intake receipt.authorization_receipt.purpose",
)
authorized_source_sha256 = _receipt_sha256(
authorization["authorized_source_sha256"],
label="intake receipt.authorization_receipt.authorized_source_sha256",
)
if authorized_source_sha256 != source["sha256"]:
raise ContractValidationError(
"intake receipt authorization is not bound to the source digest"
)
_exact_text_list(
authorization["permitted_actions"],
expected=PERMITTED_ACTIONS,
label="intake receipt.authorization_receipt.permitted_actions",
)
_exact_text_list(
authorization["prohibited_actions"],
expected=PROHIBITED_ACTIONS,
label="intake receipt.authorization_receipt.prohibited_actions",
)
_receipt_sha256(
authorization["authorization_content_sha256"],
label="intake receipt.authorization_receipt.authorization_content_sha256",
)
_require_constant(
authorization["assurance"],
expected="declaration_bound_to_exact_source_not_independently_verified",
label="intake receipt.authorization_receipt.assurance",
)
privacy = _mapping(
receipt["privacy_boundary"],
label="intake receipt.privacy_boundary",
)
_exact_fields(
privacy,
required=frozenset(
{
"codex_context_acknowledged",
"automatic_anonymization_claimed",
"clara_external_recipient_added",
"declared_raw_and_row_level_storage",
"repository_recording_policy",
"storage_location_check",
}
),
label="intake receipt.privacy_boundary",
)
for field, expected in (
("codex_context_acknowledged", True),
("automatic_anonymization_claimed", False),
("clara_external_recipient_added", False),
("declared_raw_and_row_level_storage", "local_run_root_only"),
("repository_recording_policy", "sanitized_summary_and_receipts_only"),
):
_require_constant(
privacy[field],
expected=expected,
label=f"intake receipt.privacy_boundary.{field}",
)
location = _mapping(
privacy["storage_location_check"],
label="intake receipt.privacy_boundary.storage_location_check",
)
_exact_fields(
location,
required=frozenset(
{
"status",
"source_relation",
"intake_relation",
"resolved_path_containment_enforced",
"git_ignore_required_inside_repository",
}
),
label="intake receipt.privacy_boundary.storage_location_check",
)
_require_constant(
location["status"],
expected="passed_at_validation",
label="intake receipt storage status",
)
for field in ("source_relation", "intake_relation"):
relation = _text(
location[field],
label=f"intake receipt storage {field}",
)
if relation not in STORAGE_RELATIONS:
raise ContractValidationError(
f"intake receipt storage {field} is not registered"
)
for field in (
"resolved_path_containment_enforced",
"git_ignore_required_inside_repository",
):
_require_constant(
location[field],
expected=True,
label=f"intake receipt storage {field}",
)
deidentification = _mapping(
receipt["deidentification_review"],
label="intake receipt.deidentification_review",
)
_exact_fields(
deidentification,
required=frozenset(
{
"status",
"reidentification_risk_review_status",
"review_content_sha256",
"assurance",
"basis_recorded",
}
),
label="intake receipt.deidentification_review",
)
deidentification_status = _text(
deidentification["status"],
label="intake receipt.deidentification_review.status",
)
risk_status = _text(
deidentification["reidentification_risk_review_status"],
label=(
"intake receipt.deidentification_review."
"reidentification_risk_review_status"
),
)
allowed_deidentification_pairs = {
("not_applicable", "not_applicable"),
("reviewed", "reviewed"),
}
if (deidentification_status, risk_status) not in allowed_deidentification_pairs:
raise ContractValidationError(
"intake receipt de-identification statuses must match"
)
if data_classification == "anonymized_real" and (
deidentification_status,
risk_status,
) != ("reviewed", "reviewed"):
raise ContractValidationError(
"anonymized intake receipt requires reviewed de-identification"
)
_receipt_sha256(
deidentification["review_content_sha256"],
label="intake receipt.deidentification_review.review_content_sha256",
)
_require_constant(
deidentification["assurance"],
expected="review_presence_only_not_anonymization_certification",
label="intake receipt.deidentification_review.assurance",
)
_require_constant(
deidentification["basis_recorded"],
expected=True,
label="intake receipt.deidentification_review.basis_recorded",
)
semantic = _mapping(
receipt["semantic_review"],
label="intake receipt.semantic_review",
)
_exact_fields(
semantic,
required=frozenset(
{
"status",
"required_review_count",
"automatic_mapping_allowed",
"unresolved_blocking_issues_block_preparation",
}
),
label="intake receipt.semantic_review",
)
for field, expected in (
("status", "not_assessed"),
("required_review_count", len(REQUIRED_SEMANTIC_REVIEWS)),
("automatic_mapping_allowed", False),
("unresolved_blocking_issues_block_preparation", True),
):
_require_constant(
semantic[field],
expected=expected,
label=f"intake receipt.semantic_review.{field}",
)
eligibility = _mapping(
receipt["eligibility"],
label="intake receipt.eligibility",
)
_exact_fields(
eligibility,
required=frozenset(
{
"status",
"purpose",
"publication_status",
"report_ready",
"execution_revalidation_required",
"does_not_establish",
}
),
label="intake receipt.eligibility",
)
for field, expected in (
("status", "declared_boundary_passed_for_local_pilot_intake"),
("purpose", PURPOSE),
("publication_status", "withheld"),
("report_ready", False),
("execution_revalidation_required", True),
):
_require_constant(
eligibility[field],
expected=expected,
label=f"intake receipt.eligibility.{field}",
)
_exact_text_list(
eligibility["does_not_establish"],
expected=does_not_establish,
label="intake receipt.eligibility.does_not_establish",
)
validator = _mapping(
receipt["validator"],
label="intake receipt.validator",
)
_exact_fields(
validator,
required=frozenset(
{
"dependency_sha256",
"validator_id",
"validator_version",
"implementation_sha256",
"mode",
}
),
label="intake receipt.validator",
)
_require_constant(
validator["validator_id"],
expected=validator_id,
label="intake receipt.validator.validator_id",
)
_require_constant(
validator["validator_version"],
expected=validator_version,
label="intake receipt.validator.validator_version",
)
implementation_sha256 = _receipt_sha256(
validator["implementation_sha256"],
label="intake receipt.validator.implementation_sha256",
)
if implementation_sha256 != file_sha256(Path(__file__).resolve()):
raise ContractValidationError(
"intake receipt validator implementation is not current"
)
_validate_dependency_sha256(
validator["dependency_sha256"],
label="intake receipt.validator.dependency_sha256",
dependency_paths=dependency_paths,
)
_require_constant(
validator["mode"],
expected="deterministic_mechanical",
label="intake receipt.validator.mode",
)
return dict(receipt)
def validate_real_data_pilot_intake_receipt(value: Any) -> dict[str, Any]:
"""Validate one frozen v1 trial-balance intake receipt."""
return _validate_real_data_pilot_intake_receipt(
value,
receipt_schema=INTAKE_RECEIPT_SCHEMA,
source_kinds=frozenset({SOURCE_KIND}),
does_not_establish=DOES_NOT_ESTABLISH,
validator_id=VALIDATOR_ID,
validator_version=VALIDATOR_VERSION,
dependency_paths=INTAKE_VALIDATOR_DEPENDENCY_PATHS,
)
def validate_real_data_pilot_intake_receipt_v2(value: Any) -> dict[str, Any]:
"""Validate one v2 trial-balance or general-ledger intake receipt."""
return _validate_real_data_pilot_intake_receipt(
value,
receipt_schema=INTAKE_RECEIPT_SCHEMA_V2,
source_kinds=SOURCE_KINDS_V2,
does_not_establish=DOES_NOT_ESTABLISH_V2,
validator_id=VALIDATOR_ID_V2,
validator_version=VALIDATOR_VERSION_V2,
dependency_paths=INTAKE_VALIDATOR_DEPENDENCY_PATHS_V2,
)
def _validate_real_data_pilot_intake(
intake_path: Path,
source_path: Path,
*,
as_of_date: str,
local_run_root: Path,
repository_root: Path,
intake_schema: str,
receipt_schema: str,
source_kinds: frozenset[str],
does_not_establish: tuple[str, ...],
validator_id: str,
validator_version: str,
dependency_paths: Mapping[str, Path],
) -> dict[str, Any]:
"""Return a versioned source-bound receipt without reading accounting meaning."""
run_root, repo_root = resolve_pilot_storage_roots(
local_run_root=local_run_root,
repository_root=repository_root,
)
intake_path, intake_relation = validate_pilot_local_run_path(
intake_path,
local_run_root=run_root,
repository_root=repo_root,
label="intake path",
)
source_path, source_relation = validate_pilot_local_run_path(
source_path,
local_run_root=run_root,
repository_root=repo_root,
label="source path",
)
if not intake_path.is_file():
raise ContractValidationError("intake path must identify one existing file")
intake, _, intake_contract_sha256 = strict_json_snapshot_beneath(
intake_path,
root=run_root,
)
_exact_fields(
intake,
required=frozenset(
{
"schema_version",
"pilot_id",
"purpose",
"source",
"authorization",
"privacy",
"deidentification_review",
"semantic_review_plan",
"publication_status",
"report_ready",
}
),
label="intake",
)
_require_constant(
intake["schema_version"],
expected=intake_schema,
label="intake.schema_version",
)
pilot_id = _opaque_identifier(
intake["pilot_id"],
pattern=PILOT_ID_PATTERN,
label="intake.pilot_id",
)
_require_constant(intake["purpose"], expected=PURPOSE, label="intake.purpose")
source = _validate_source_declaration(
intake["source"],
source_kinds=source_kinds,
)
as_of = _iso_date(as_of_date, label="as_of_date")
authorization = _validate_authorization(
intake["authorization"],
source_sha256=source["sha256"],
as_of=as_of,
)
privacy = _validate_privacy(intake["privacy"])
deidentification_review = _validate_deidentification_review(
intake["deidentification_review"],
data_classification=source["data_classification"],
)
semantic_review = _validate_semantic_review_plan(intake["semantic_review_plan"])
_require_constant(
intake["publication_status"],
expected="withheld",
label="intake.publication_status",
)
_require_constant(
intake["report_ready"],
expected=False,
label="intake.report_ready",
)
source_receipt = _bind_source_file(
source,
source_path=source_path,
local_run_root=run_root,
)
privacy_boundary = {
**privacy,
"storage_location_check": {
"status": "passed_at_validation",
"source_relation": source_relation,
"intake_relation": intake_relation,
"resolved_path_containment_enforced": True,
"git_ignore_required_inside_repository": True,
},
}
receipt = {
"schema_version": receipt_schema,
"pilot_id": pilot_id,
"validation_date": as_of.isoformat(),
"intake_contract_sha256": intake_contract_sha256,
"source_receipt": source_receipt,
"authorization_receipt": authorization,
"privacy_boundary": privacy_boundary,
"deidentification_review": deidentification_review,
"semantic_review": semantic_review,
"eligibility": {
"status": "declared_boundary_passed_for_local_pilot_intake",
"purpose": PURPOSE,
"publication_status": "withheld",
"report_ready": False,
"execution_revalidation_required": True,
"does_not_establish": list(does_not_establish),
},
"validator": {
"dependency_sha256": _current_dependency_sha256(dependency_paths),
"validator_id": validator_id,
"validator_version": validator_version,
"implementation_sha256": file_sha256(Path(__file__).resolve()),
"mode": "deterministic_mechanical",
},
}
return _validate_real_data_pilot_intake_receipt(
receipt,
receipt_schema=receipt_schema,
source_kinds=source_kinds,
does_not_establish=does_not_establish,
validator_id=validator_id,
validator_version=validator_version,
dependency_paths=dependency_paths,
)
def validate_real_data_pilot_intake(
intake_path: Path,
source_path: Path,
*,
as_of_date: str,
local_run_root: Path,
repository_root: Path,
) -> dict[str, Any]:
"""Return one frozen v1 trial-balance intake receipt."""
return _validate_real_data_pilot_intake(
intake_path,
source_path,
as_of_date=as_of_date,
local_run_root=local_run_root,
repository_root=repository_root,
intake_schema=INTAKE_SCHEMA,
receipt_schema=INTAKE_RECEIPT_SCHEMA,
source_kinds=frozenset({SOURCE_KIND}),
does_not_establish=DOES_NOT_ESTABLISH,
validator_id=VALIDATOR_ID,
validator_version=VALIDATOR_VERSION,
dependency_paths=INTAKE_VALIDATOR_DEPENDENCY_PATHS,
)
def validate_real_data_pilot_intake_v2(
intake_path: Path,
source_path: Path,
*,
as_of_date: str,
local_run_root: Path,
repository_root: Path,
) -> dict[str, Any]:
"""Return one v2 trial-balance or general-ledger intake receipt."""
return _validate_real_data_pilot_intake(
intake_path,
source_path,
as_of_date=as_of_date,
local_run_root=local_run_root,
repository_root=repository_root,
intake_schema=INTAKE_SCHEMA_V2,
receipt_schema=INTAKE_RECEIPT_SCHEMA_V2,
source_kinds=SOURCE_KINDS_V2,
does_not_establish=DOES_NOT_ESTABLISH_V2,
validator_id=VALIDATOR_ID_V2,
validator_version=VALIDATOR_VERSION_V2,
dependency_paths=INTAKE_VALIDATOR_DEPENDENCY_PATHS_V2,
)
def main(argv: Sequence[str] | None = None) -> int:
"""Validate one intake contract and write its sanitized receipt."""
parser = argparse.ArgumentParser(
description="Validate Clara's real-data pilot intake boundary."
)
parser.add_argument("--intake", required=True, type=Path)
parser.add_argument("--source", required=True, type=Path)
parser.add_argument("--local-run-root", required=True, type=Path)
parser.add_argument("--repository-root", required=True, type=Path)
parser.add_argument("--output", required=True, type=Path)
parser.add_argument(
"--contract-version",
choices=("v1", "v2"),
default="v1",
help="Intake contract version; v1 is trial-balance-only.",
)
args = parser.parse_args(argv)
if args.contract_version == "v2":
validate_intake = validate_real_data_pilot_intake_v2
else:
validate_intake = validate_real_data_pilot_intake
run_root, repo_root = resolve_pilot_storage_roots(
local_run_root=args.local_run_root,
repository_root=args.repository_root,
)
output_path, _ = validate_pilot_receipt_output_path(
args.output,
declared_local_run_root=args.local_run_root,
local_run_root=run_root,
repository_root=repo_root,
label="output path",
)
source_path = args.source.resolve()
intake_path = args.intake.resolve()
if output_path in {source_path, intake_path}:
raise ContractValidationError(
"output path must not equal the source or intake path"
)
if not output_path.parent.is_dir():
raise ContractValidationError(
"output path parent must be an existing directory"
)
with pinned_pilot_receipt_output(
output_path,
local_run_root=run_root,
) as pinned_output:
pinned_output.require_absent()
receipt = validate_intake(
intake_path,
source_path,
as_of_date=_current_date().isoformat(),
local_run_root=run_root,
repository_root=repo_root,
)
pinned_output.write_json(receipt)
LOGGER.info(
"Real-data pilot intake %s: %s",
receipt["pilot_id"],
receipt["eligibility"]["status"],
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
SHA-256: 74f9277e63917895f673868bd5ee596660375b7df6105915f7922a4278490089