837 lines
28 KiB
Python
837 lines
28 KiB
Python
"""Strict restored-audit verification at reconciliation, resume and READY boundaries."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
from sqlalchemy import Column, Integer, MetaData, Table, create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from modelforge_api.domain.audit import AUDIT_EMPTY_LEGACY_PREFIX_SEAL
|
|
from modelforge_api.domain.recovery import (
|
|
RESTORE_PHASE_ORDER,
|
|
RecoveryFailureCode,
|
|
RestoreAdvanceRequest,
|
|
RestoreState,
|
|
)
|
|
from modelforge_api.persistence.models import Base
|
|
from modelforge_api.services.audit import (
|
|
AUDIT_CURRENT_HASH_FORMAT,
|
|
AUDIT_HASH_FORMAT_V1,
|
|
AuditChainCheckpoint,
|
|
AuditEventRecord,
|
|
canonical_audit_payload_and_hash,
|
|
canonical_audit_payload_text_and_hash,
|
|
legacy_audit_prefix_seal,
|
|
normalise_audit_timestamp,
|
|
)
|
|
from modelforge_api.services.recovery import RecoveryError, RecoveryService
|
|
from modelforge_api.services.recovery_fingerprint import (
|
|
CURRENT_TRUTH_TABLES,
|
|
FINGERPRINT_VERSION,
|
|
LEGACY_FINGERPRINT_VERSION,
|
|
MAX_ROWS_PER_TABLE,
|
|
_canonical_row,
|
|
_table_digest,
|
|
audit_fingerprint_differences_are_compatible,
|
|
classified_tables,
|
|
fingerprint_compatibility,
|
|
fingerprint_session,
|
|
)
|
|
from modelforge_api.services.recovery_postgres import (
|
|
CommandResult,
|
|
PostgresEngine,
|
|
PostgresTarget,
|
|
PostgresToolError,
|
|
)
|
|
from modelforge_api.settings import Settings
|
|
|
|
TARGET = PostgresTarget("postgres", 5432, "restored", "modelforge", None)
|
|
|
|
|
|
@pytest.fixture
|
|
def session() -> Session:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as value:
|
|
yield value
|
|
|
|
|
|
def _service(session: Session, tmp_path: Path) -> RecoveryService:
|
|
key = base64.b64encode(b"modelforge-rc-audit-test-key!!"[:32]).decode()
|
|
return RecoveryService(
|
|
session,
|
|
Settings(
|
|
backup_root=tmp_path / "backups",
|
|
backup_restore_root=tmp_path / "restore",
|
|
backup_encryption_key=key,
|
|
config_root=tmp_path / "config",
|
|
database_url="postgresql+psycopg://modelforge:test@postgres:5432/modelforge",
|
|
),
|
|
)
|
|
|
|
|
|
def _mixed_records() -> tuple[list[AuditEventRecord], AuditChainCheckpoint]:
|
|
legacy_id = uuid.UUID("11111111-1111-1111-1111-111111111111")
|
|
legacy_time = datetime(2026, 8, 28, 20, 0, 0, 123456, tzinfo=UTC)
|
|
legacy_payload, legacy_hash = canonical_audit_payload_and_hash(
|
|
correlation_id="legacy-request",
|
|
actor_type="operator",
|
|
actor_id="legacy-operator",
|
|
action="LEGACY_APPROVAL",
|
|
resource_type="revision",
|
|
resource_id="revision-1",
|
|
outcome="success",
|
|
details={"approved": True},
|
|
previous_event_hash=None,
|
|
hash_format=AUDIT_HASH_FORMAT_V1,
|
|
)
|
|
legacy = AuditEventRecord(
|
|
id=legacy_id,
|
|
sequence=1,
|
|
occurred_at=legacy_time,
|
|
event_hash=legacy_hash,
|
|
hash_format=AUDIT_HASH_FORMAT_V1,
|
|
**legacy_payload,
|
|
)
|
|
|
|
v2_id = uuid.UUID("22222222-2222-2222-2222-222222222222")
|
|
v2_time = legacy_time + timedelta(seconds=1)
|
|
v2_payload, v2_canonical_payload, v2_hash = canonical_audit_payload_text_and_hash(
|
|
correlation_id="v2-request",
|
|
actor_type="operator",
|
|
actor_id="v2-operator",
|
|
action="V2_APPROVAL",
|
|
resource_type="revision",
|
|
resource_id="revision-2",
|
|
outcome="success",
|
|
details={"approved": True},
|
|
previous_event_hash=legacy_hash,
|
|
hash_format=AUDIT_CURRENT_HASH_FORMAT,
|
|
event_id=v2_id,
|
|
occurred_at=v2_time,
|
|
)
|
|
v2 = AuditEventRecord(
|
|
id=v2_id,
|
|
sequence=2,
|
|
occurred_at=v2_time,
|
|
event_hash=v2_hash,
|
|
hash_format=AUDIT_CURRENT_HASH_FORMAT,
|
|
canonical_payload=v2_canonical_payload,
|
|
**v2_payload,
|
|
)
|
|
checkpoint = AuditChainCheckpoint(
|
|
singleton_id=1,
|
|
event_count=2,
|
|
last_sequence=2,
|
|
last_event_hash=v2_hash,
|
|
hash_format=AUDIT_CURRENT_HASH_FORMAT,
|
|
v2_start_sequence=2,
|
|
legacy_prefix_count=1,
|
|
legacy_prefix_seal=legacy_audit_prefix_seal([legacy]),
|
|
)
|
|
return [legacy, v2], checkpoint
|
|
|
|
|
|
def _event_row(event: AuditEventRecord) -> dict[str, str]:
|
|
return {
|
|
"id": str(event.id),
|
|
"sequence": str(event.sequence),
|
|
"occurred_at": normalise_audit_timestamp(event.occurred_at),
|
|
"correlation_id": event.correlation_id,
|
|
"actor_type": event.actor_type,
|
|
"actor_id": event.actor_id,
|
|
"action": event.action,
|
|
"resource_type": event.resource_type,
|
|
"resource_id": event.resource_id or "",
|
|
"resource_id_is_null": "t" if event.resource_id is None else "f",
|
|
"outcome": event.outcome,
|
|
"details": json.dumps(event.details, sort_keys=True, separators=(",", ":")),
|
|
"previous_event_hash": event.previous_event_hash or "",
|
|
"previous_event_hash_is_null": "t" if event.previous_event_hash is None else "f",
|
|
"event_hash": event.event_hash,
|
|
"hash_format": event.hash_format,
|
|
"canonical_payload": event.canonical_payload or "",
|
|
"canonical_payload_is_null": "t" if event.canonical_payload is None else "f",
|
|
}
|
|
|
|
|
|
def _head_row(checkpoint: AuditChainCheckpoint) -> dict[str, str]:
|
|
return {
|
|
"singleton_id": str(checkpoint.singleton_id),
|
|
"event_count": str(checkpoint.event_count),
|
|
"last_sequence": str(checkpoint.last_sequence),
|
|
"last_event_hash": checkpoint.last_event_hash or "",
|
|
"last_event_hash_is_null": "t" if checkpoint.last_event_hash is None else "f",
|
|
"hash_format": checkpoint.hash_format,
|
|
"v2_start_sequence": str(checkpoint.v2_start_sequence),
|
|
"legacy_prefix_count": str(checkpoint.legacy_prefix_count),
|
|
"legacy_prefix_seal": checkpoint.legacy_prefix_seal,
|
|
}
|
|
|
|
|
|
class SnapshotEngine:
|
|
def __init__(
|
|
self,
|
|
events: list[dict[str, str]],
|
|
checkpoint: dict[str, str],
|
|
*,
|
|
marker_count: str = "0",
|
|
) -> None:
|
|
self.events = events
|
|
self.checkpoint = checkpoint
|
|
self.marker_count = marker_count
|
|
self.operations: list[str] = []
|
|
|
|
def query_rows(
|
|
self,
|
|
_target: PostgresTarget,
|
|
sql: str,
|
|
*,
|
|
max_rows: int = 1000,
|
|
) -> list[dict[str, str]]:
|
|
self.operations.append("query_rows")
|
|
if "from audit_chain_heads" in sql:
|
|
return [copy.deepcopy(self.checkpoint)]
|
|
if "count(distinct id)" in sql:
|
|
return [
|
|
{
|
|
"event_count": str(len(self.events)),
|
|
"distinct_event_ids": str(
|
|
len({row["id"] for row in self.events})
|
|
),
|
|
"distinct_sequences": str(
|
|
len({row["sequence"] for row in self.events})
|
|
),
|
|
}
|
|
]
|
|
match = re.search(
|
|
r"sequence > (-?\d+) or \(sequence = -?\d+ and "
|
|
r"id > '([0-9a-f-]+)'::uuid\)",
|
|
sql,
|
|
)
|
|
after = int(match.group(1)) if match is not None else None
|
|
after_id = uuid.UUID(match.group(2)) if match is not None else None
|
|
return [
|
|
copy.deepcopy(row)
|
|
for row in self.events
|
|
if after is None
|
|
or (int(row["sequence"]), uuid.UUID(row["id"])) > (after, after_id)
|
|
][:max_rows]
|
|
|
|
def scalar(self, _target: PostgresTarget, _sql: str) -> str:
|
|
self.operations.append("scalar")
|
|
return self.marker_count
|
|
|
|
|
|
def _engine() -> SnapshotEngine:
|
|
records, checkpoint = _mixed_records()
|
|
return SnapshotEngine([_event_row(event) for event in records], _head_row(checkpoint))
|
|
|
|
|
|
def test_recovery_strictly_accepts_a_valid_mixed_legacy_v2_chain(
|
|
session: Session, tmp_path: Path
|
|
) -> None:
|
|
subject = _service(session, tmp_path)
|
|
subject.engine = _engine() # type: ignore[assignment]
|
|
|
|
checkpoint = subject._verify_restored_audit_chain(TARGET)
|
|
|
|
assert checkpoint.event_count == 2
|
|
assert checkpoint.v2_start_sequence == 2
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("event_index", "field", "replacement"),
|
|
[
|
|
(0, "id", "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"),
|
|
(0, "occurred_at", "2026-08-29T20:00:00.123456Z"),
|
|
(1, "id", "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"),
|
|
(1, "occurred_at", "2026-08-29T20:00:01.123456Z"),
|
|
],
|
|
)
|
|
def test_recovery_rejects_id_and_timestamp_tamper_in_legacy_and_v2(
|
|
session: Session,
|
|
tmp_path: Path,
|
|
event_index: int,
|
|
field: str,
|
|
replacement: str,
|
|
) -> None:
|
|
engine = _engine()
|
|
engine.events[event_index][field] = replacement
|
|
subject = _service(session, tmp_path)
|
|
subject.engine = engine # type: ignore[assignment]
|
|
|
|
with pytest.raises(RecoveryError) as captured:
|
|
subject._verify_restored_audit_chain(TARGET)
|
|
|
|
assert captured.value.code == RecoveryFailureCode.AUDIT_CHAIN_CORRUPT.value
|
|
|
|
|
|
@pytest.mark.parametrize("retained", [1, 0])
|
|
def test_recovery_rejects_tail_and_complete_history_deletion(
|
|
session: Session, tmp_path: Path, retained: int
|
|
) -> None:
|
|
engine = _engine()
|
|
engine.events = engine.events[:retained]
|
|
subject = _service(session, tmp_path)
|
|
subject.engine = engine # type: ignore[assignment]
|
|
|
|
with pytest.raises(RecoveryError) as captured:
|
|
subject._verify_restored_audit_chain(TARGET)
|
|
|
|
assert captured.value.code == RecoveryFailureCode.AUDIT_CHAIN_CORRUPT.value
|
|
assert "checkpoint" in str(captured.value.details).lower()
|
|
|
|
|
|
def test_corruption_blocks_before_already_applied_marker_lookup(
|
|
session: Session, tmp_path: Path
|
|
) -> None:
|
|
engine = _engine()
|
|
engine.events.pop()
|
|
subject = _service(session, tmp_path)
|
|
subject.engine = engine # type: ignore[assignment]
|
|
record = SimpleNamespace(id=uuid.uuid4())
|
|
plan = SimpleNamespace(
|
|
database_destination="postgresql://modelforge:test@postgres:5432/restored",
|
|
backup_set=SimpleNamespace(backup_id="backup-1"),
|
|
)
|
|
|
|
with pytest.raises(RecoveryError) as captured:
|
|
subject._reconcile_restored(record, plan)
|
|
|
|
assert captured.value.code == RecoveryFailureCode.AUDIT_CHAIN_CORRUPT.value
|
|
assert "scalar" not in engine.operations
|
|
|
|
|
|
def test_valid_resume_checks_chain_before_returning_already_applied(
|
|
session: Session, tmp_path: Path
|
|
) -> None:
|
|
engine = _engine()
|
|
engine.marker_count = "1"
|
|
subject = _service(session, tmp_path)
|
|
subject.engine = engine # type: ignore[assignment]
|
|
record = SimpleNamespace(id=uuid.uuid4())
|
|
plan = SimpleNamespace(
|
|
database_destination="postgresql://modelforge:test@postgres:5432/restored",
|
|
backup_set=SimpleNamespace(backup_id="backup-1"),
|
|
)
|
|
|
|
result = subject._reconcile_restored(record, plan)
|
|
|
|
assert result["reconciliation"] == "ALREADY_APPLIED"
|
|
assert engine.operations.index("query_rows") < engine.operations.index("scalar")
|
|
|
|
|
|
def test_corruption_is_an_independent_ready_gate_before_fingerprints(
|
|
session: Session, tmp_path: Path
|
|
) -> None:
|
|
engine = _engine()
|
|
engine.events.clear()
|
|
subject = _service(session, tmp_path)
|
|
subject.engine = engine # type: ignore[assignment]
|
|
record = SimpleNamespace()
|
|
plan = SimpleNamespace(
|
|
database_destination="postgresql://modelforge:test@postgres:5432/restored"
|
|
)
|
|
|
|
with pytest.raises(RecoveryError) as captured:
|
|
subject._validate_restored(record, plan)
|
|
|
|
assert captured.value.code == RecoveryFailureCode.AUDIT_CHAIN_CORRUPT.value
|
|
|
|
|
|
def test_ready_resume_reverifies_even_when_every_phase_duration_is_present(
|
|
session: Session, tmp_path: Path
|
|
) -> None:
|
|
subject = _service(session, tmp_path)
|
|
_records, checkpoint = _mixed_records()
|
|
verifier = Mock(return_value=checkpoint)
|
|
subject._verify_restored_audit_chain = verifier # type: ignore[method-assign]
|
|
subject._journal = Mock() # type: ignore[method-assign]
|
|
subject.audit = SimpleNamespace(write=Mock()) # type: ignore[assignment]
|
|
|
|
plan = SimpleNamespace(
|
|
database_destination="postgresql://modelforge:test@postgres:5432/restored",
|
|
backup_set=SimpleNamespace(backup_id="backup-1"),
|
|
state="PREFLIGHT_PASSED",
|
|
)
|
|
record = SimpleNamespace(
|
|
id=uuid.uuid4(),
|
|
state=RestoreState.VALIDATING.value,
|
|
plan=plan,
|
|
phase_durations={phase.value: 0.1 for phase in RESTORE_PHASE_ORDER},
|
|
rto_seconds=None,
|
|
rpo_seconds=0.0,
|
|
ready_at=None,
|
|
)
|
|
subject._operation_row = lambda _operation_id: record # type: ignore[method-assign]
|
|
subject._operation_response = lambda value: value # type: ignore[method-assign]
|
|
|
|
result = subject.advance_restore(
|
|
record.id,
|
|
RestoreAdvanceRequest(actor="operator", reason="resume final READY control"),
|
|
)
|
|
|
|
assert result.state == RestoreState.READY.value
|
|
verifier.assert_called_once()
|
|
verified_target = verifier.call_args.args[0]
|
|
assert (verified_target.host, verified_target.database, verified_target.user) == (
|
|
TARGET.host,
|
|
TARGET.database,
|
|
TARGET.user,
|
|
)
|
|
assert record.phase_durations[RestoreState.READY.value] >= 0
|
|
|
|
|
|
def _v2_chain_rows(
|
|
count: int,
|
|
*,
|
|
duplicate_id_at: int | None = None,
|
|
duplicate_sequence_at: int | None = None,
|
|
) -> tuple[list[dict[str, str]], AuditChainCheckpoint]:
|
|
rows: list[dict[str, str]] = []
|
|
previous_hash: str | None = None
|
|
started = datetime(2026, 8, 30, 12, 0, tzinfo=UTC)
|
|
for index in range(1, count + 1):
|
|
event_id = uuid.UUID(
|
|
int=(index - 1 if duplicate_id_at == index else index)
|
|
)
|
|
occurred_at = started + timedelta(microseconds=index)
|
|
payload, canonical_payload, event_hash = canonical_audit_payload_text_and_hash(
|
|
correlation_id=f"request-{index}",
|
|
actor_type="operator",
|
|
actor_id="pagination-control",
|
|
action="VALID_EVENT",
|
|
resource_type="audit-test",
|
|
resource_id=str(index),
|
|
outcome="success",
|
|
details={"index": index},
|
|
previous_event_hash=previous_hash,
|
|
hash_format=AUDIT_CURRENT_HASH_FORMAT,
|
|
event_id=event_id,
|
|
occurred_at=occurred_at,
|
|
)
|
|
record = AuditEventRecord(
|
|
id=event_id,
|
|
sequence=index,
|
|
occurred_at=occurred_at,
|
|
event_hash=event_hash,
|
|
hash_format=AUDIT_CURRENT_HASH_FORMAT,
|
|
canonical_payload=canonical_payload,
|
|
**payload,
|
|
)
|
|
row = _event_row(record)
|
|
if duplicate_sequence_at == index:
|
|
row["sequence"] = str(index - 1)
|
|
rows.append(row)
|
|
previous_hash = event_hash
|
|
checkpoint = AuditChainCheckpoint(
|
|
singleton_id=1,
|
|
event_count=count,
|
|
last_sequence=count,
|
|
last_event_hash=rows[-1]["event_hash"],
|
|
hash_format=AUDIT_CURRENT_HASH_FORMAT,
|
|
v2_start_sequence=1,
|
|
legacy_prefix_count=0,
|
|
legacy_prefix_seal=AUDIT_EMPTY_LEGACY_PREFIX_SEAL,
|
|
)
|
|
return rows, checkpoint
|
|
|
|
|
|
def test_valid_501_row_chain_crosses_the_composite_cursor_boundary(
|
|
session: Session, tmp_path: Path
|
|
) -> None:
|
|
rows, checkpoint = _v2_chain_rows(501)
|
|
engine = SnapshotEngine(rows, _head_row(checkpoint))
|
|
subject = _service(session, tmp_path)
|
|
subject.engine = engine # type: ignore[assignment]
|
|
|
|
verified = subject._verify_restored_audit_chain(TARGET)
|
|
|
|
assert verified.event_count == 501
|
|
assert engine.operations.count("query_rows") == 4
|
|
|
|
|
|
def test_composite_cursor_does_not_skip_the_501st_duplicate_sequence(
|
|
session: Session, tmp_path: Path
|
|
) -> None:
|
|
rows, checkpoint = _v2_chain_rows(501, duplicate_sequence_at=501)
|
|
engine = SnapshotEngine(rows, _head_row(checkpoint))
|
|
subject = _service(session, tmp_path)
|
|
subject.engine = engine # type: ignore[assignment]
|
|
|
|
with pytest.raises(RecoveryError) as captured:
|
|
subject._verify_restored_audit_chain(TARGET)
|
|
|
|
assert captured.value.code == RecoveryFailureCode.AUDIT_CHAIN_CORRUPT.value
|
|
assert engine.operations.count("query_rows") == 4 # head, identity counts, both pages
|
|
assert any(
|
|
"not unique" in violation or "expected 501" in violation
|
|
for violation in captured.value.details["violations"]
|
|
)
|
|
|
|
|
|
def test_strict_recovery_rejects_a_valid_hash_chain_with_duplicate_event_uuid(
|
|
session: Session, tmp_path: Path
|
|
) -> None:
|
|
rows, checkpoint = _v2_chain_rows(501, duplicate_id_at=501)
|
|
engine = SnapshotEngine(rows, _head_row(checkpoint))
|
|
subject = _service(session, tmp_path)
|
|
subject.engine = engine # type: ignore[assignment]
|
|
|
|
with pytest.raises(RecoveryError) as captured:
|
|
subject._verify_restored_audit_chain(TARGET)
|
|
|
|
assert captured.value.code == RecoveryFailureCode.AUDIT_CHAIN_CORRUPT.value
|
|
assert engine.operations.count("query_rows") == 4
|
|
violations = captured.value.details["violations"]
|
|
assert any("UUID" in violation and "unique" in violation for violation in violations)
|
|
assert any("UUID" in violation and "duplicated" in violation for violation in violations)
|
|
|
|
|
|
def _fingerprint(
|
|
version: str,
|
|
*,
|
|
include_checkpoint: bool,
|
|
) -> dict[str, object]:
|
|
table_names = set(classified_tables())
|
|
if not include_checkpoint:
|
|
table_names.remove("audit_chain_heads")
|
|
tables: dict[str, object] = {
|
|
name: {
|
|
"row_count": 0,
|
|
"digest": hashlib.sha256(name.encode()).hexdigest(),
|
|
"status": "COMPLETE",
|
|
"redacted_columns": [],
|
|
}
|
|
for name in sorted(table_names)
|
|
}
|
|
tables["audit_events"] = {
|
|
"row_count": 12,
|
|
"digest": "a" * 64,
|
|
"status": "COMPLETE",
|
|
"redacted_columns": [],
|
|
}
|
|
if include_checkpoint:
|
|
tables["audit_chain_heads"] = {
|
|
"row_count": 1,
|
|
"digest": "b" * 64,
|
|
"status": "COMPLETE",
|
|
"redacted_columns": [],
|
|
}
|
|
return {
|
|
"version": version,
|
|
"tables": tables,
|
|
"digest": hashlib.sha256(version.encode()).hexdigest(),
|
|
"groups": {},
|
|
"physical_tables": sorted(table_names | CURRENT_TRUTH_TABLES),
|
|
"unexpected_tables": [],
|
|
}
|
|
|
|
|
|
def _with_legitimate_reconciliation_delta(
|
|
source: dict[str, object], *, introduce_checkpoint: bool
|
|
) -> dict[str, object]:
|
|
restored = copy.deepcopy(source)
|
|
restored["version"] = FINGERPRINT_VERSION
|
|
restored["digest"] = "f" * 64
|
|
restored["unexpected_tables"] = []
|
|
restored["physical_tables"] = sorted(classified_tables() | CURRENT_TRUTH_TABLES)
|
|
tables = restored["tables"]
|
|
assert isinstance(tables, dict)
|
|
tables["audit_events"] = {
|
|
"row_count": 13,
|
|
"digest": "c" * 64,
|
|
"status": "COMPLETE",
|
|
"redacted_columns": [],
|
|
}
|
|
tables["audit_chain_heads"] = {
|
|
"row_count": 1,
|
|
"digest": "d" * 64,
|
|
"status": "COMPLETE",
|
|
"redacted_columns": [],
|
|
}
|
|
if not introduce_checkpoint:
|
|
tables["operational_alerts"] = {
|
|
"row_count": 0,
|
|
"digest": "e" * 64,
|
|
"status": "COMPLETE",
|
|
"redacted_columns": [],
|
|
}
|
|
return restored
|
|
|
|
|
|
def test_schema_0022_fingerprint_allows_exact_checkpoint_introduction_only() -> None:
|
|
source = _fingerprint(
|
|
LEGACY_FINGERPRINT_VERSION,
|
|
include_checkpoint=False,
|
|
)
|
|
source.pop("unexpected_tables") # m15.1 predates explicit physical-table reporting
|
|
source.pop("physical_tables")
|
|
restored = _with_legitimate_reconciliation_delta(
|
|
source, introduce_checkpoint=True
|
|
)
|
|
contract = fingerprint_compatibility(
|
|
source, restored, source_schema_revision="20260828_0022"
|
|
)
|
|
|
|
assert contract.compatible is True
|
|
assert contract.mode == "SCHEMA_0022_TO_0024"
|
|
assert (
|
|
audit_fingerprint_differences_are_compatible(source, restored, contract)
|
|
is True
|
|
)
|
|
|
|
|
|
def test_schema_0024_fingerprint_allows_only_exact_reconciliation_deltas() -> None:
|
|
source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
|
|
restored = _with_legitimate_reconciliation_delta(
|
|
source, introduce_checkpoint=False
|
|
)
|
|
contract = fingerprint_compatibility(
|
|
source, restored, source_schema_revision="20260830_0024"
|
|
)
|
|
|
|
assert contract.compatible is True
|
|
assert contract.mode == "CURRENT_STRICT"
|
|
assert (
|
|
audit_fingerprint_differences_are_compatible(source, restored, contract)
|
|
is True
|
|
)
|
|
|
|
|
|
def test_current_fingerprint_missing_its_checkpoint_remains_strictly_incompatible() -> None:
|
|
source = _fingerprint(
|
|
FINGERPRINT_VERSION,
|
|
include_checkpoint=False,
|
|
)
|
|
restored = _with_legitimate_reconciliation_delta(
|
|
source, introduce_checkpoint=True
|
|
)
|
|
contract = fingerprint_compatibility(
|
|
source, restored, source_schema_revision="20260830_0024"
|
|
)
|
|
|
|
assert contract.compatible is False
|
|
assert contract.mode == "INCOMPATIBLE"
|
|
assert (
|
|
audit_fingerprint_differences_are_compatible(source, restored, contract)
|
|
is False
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("attack", ["missing_models", "attacker_table"])
|
|
def test_fingerprint_contract_rejects_sparse_or_unexpected_tables(attack: str) -> None:
|
|
source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
|
|
restored = _with_legitimate_reconciliation_delta(
|
|
source, introduce_checkpoint=False
|
|
)
|
|
tables = source["tables"]
|
|
assert isinstance(tables, dict)
|
|
if attack == "missing_models":
|
|
tables.pop("models")
|
|
else:
|
|
tables["attacker_shadow"] = {
|
|
"row_count": 1,
|
|
"digest": "9" * 64,
|
|
"status": "COMPLETE",
|
|
"redacted_columns": [],
|
|
}
|
|
|
|
contract = fingerprint_compatibility(
|
|
source, restored, source_schema_revision="20260830_0024"
|
|
)
|
|
|
|
assert contract.compatible is False
|
|
assert (
|
|
audit_fingerprint_differences_are_compatible(source, restored, contract)
|
|
is False
|
|
)
|
|
|
|
|
|
def test_semantic_fingerprint_exposes_an_unexpected_physical_table(
|
|
session: Session,
|
|
) -> None:
|
|
source = fingerprint_session(session)
|
|
session.connection().exec_driver_sql(
|
|
"CREATE TABLE attacker_shadow (payload TEXT NOT NULL)"
|
|
)
|
|
restored = fingerprint_session(session)
|
|
|
|
assert source["unexpected_tables"] == []
|
|
assert restored["unexpected_tables"] == ["attacker_shadow"]
|
|
contract = fingerprint_compatibility(
|
|
source, restored, source_schema_revision="20260830_0024"
|
|
)
|
|
assert contract.compatible is False
|
|
|
|
|
|
def test_fingerprint_rejects_large_non_audit_row_loss() -> None:
|
|
source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
|
|
source_tables = source["tables"]
|
|
assert isinstance(source_tables, dict)
|
|
source_tables["projects"] = {
|
|
"row_count": 20,
|
|
"digest": "1" * 64,
|
|
"status": "COMPLETE",
|
|
"redacted_columns": [],
|
|
}
|
|
restored = _with_legitimate_reconciliation_delta(
|
|
source, introduce_checkpoint=False
|
|
)
|
|
restored_tables = restored["tables"]
|
|
assert isinstance(restored_tables, dict)
|
|
restored_tables["projects"] = {
|
|
"row_count": 1,
|
|
"digest": "2" * 64,
|
|
"status": "COMPLETE",
|
|
"redacted_columns": [],
|
|
}
|
|
contract = fingerprint_compatibility(
|
|
source, restored, source_schema_revision="20260830_0024"
|
|
)
|
|
|
|
assert contract.compatible is True
|
|
assert (
|
|
audit_fingerprint_differences_are_compatible(source, restored, contract)
|
|
is False
|
|
)
|
|
|
|
|
|
def test_fingerprint_rejects_same_count_different_audit_digest() -> None:
|
|
source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
|
|
restored = _with_legitimate_reconciliation_delta(
|
|
source, introduce_checkpoint=False
|
|
)
|
|
restored_tables = restored["tables"]
|
|
assert isinstance(restored_tables, dict)
|
|
restored_tables["audit_events"] = {
|
|
"row_count": 12,
|
|
"digest": "c" * 64,
|
|
"status": "COMPLETE",
|
|
"redacted_columns": [],
|
|
}
|
|
contract = fingerprint_compatibility(
|
|
source, restored, source_schema_revision="20260830_0024"
|
|
)
|
|
|
|
assert contract.compatible is True
|
|
assert (
|
|
audit_fingerprint_differences_are_compatible(source, restored, contract)
|
|
is False
|
|
)
|
|
|
|
|
|
def test_fingerprint_helper_cannot_reuse_a_contract_after_table_mutation() -> None:
|
|
source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
|
|
restored = _with_legitimate_reconciliation_delta(
|
|
source, introduce_checkpoint=False
|
|
)
|
|
contract = fingerprint_compatibility(
|
|
source, restored, source_schema_revision="20260830_0024"
|
|
)
|
|
restored_tables = restored["tables"]
|
|
assert isinstance(restored_tables, dict)
|
|
restored_tables.pop("models")
|
|
|
|
assert contract.compatible is True
|
|
assert (
|
|
audit_fingerprint_differences_are_compatible(source, restored, contract)
|
|
is False
|
|
)
|
|
|
|
|
|
def test_bounded_table_fingerprint_can_never_authorize_ready() -> None:
|
|
source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
|
|
source_tables = source["tables"]
|
|
assert isinstance(source_tables, dict)
|
|
source_tables["models"] = {
|
|
"row_count": MAX_ROWS_PER_TABLE,
|
|
"digest": "7" * 64,
|
|
"status": "BOUNDED",
|
|
"redacted_columns": [],
|
|
}
|
|
restored = _with_legitimate_reconciliation_delta(
|
|
source, introduce_checkpoint=False
|
|
)
|
|
|
|
contract = fingerprint_compatibility(
|
|
source, restored, source_schema_revision="20260830_0024"
|
|
)
|
|
assert contract.compatible is False
|
|
|
|
|
|
def test_more_than_250k_rows_is_observed_as_bounded_not_a_complete_digest() -> None:
|
|
table = Table("simulated_large_table", MetaData(), Column("value", Integer()))
|
|
|
|
class SimulatedRows:
|
|
def yield_per(self, _size: int) -> Any:
|
|
for value in range(MAX_ROWS_PER_TABLE + 1):
|
|
yield (value,)
|
|
|
|
class SimulatedConnection:
|
|
def execute(self, _statement: Any) -> SimulatedRows:
|
|
return SimulatedRows()
|
|
|
|
entry = _table_digest(SimulatedConnection(), table) # type: ignore[arg-type]
|
|
|
|
assert entry["row_count"] == MAX_ROWS_PER_TABLE
|
|
assert entry["status"] == "BOUNDED"
|
|
|
|
|
|
def test_length_prefixed_typed_rows_have_no_delimiter_or_null_type_collision() -> None:
|
|
assert _canonical_row(("a\x1fb", "c")) != _canonical_row(("a", "b\x1fc"))
|
|
assert _canonical_row((None,)) != _canonical_row(("\x00",))
|
|
assert _canonical_row((1,)) != _canonical_row(("1",))
|
|
|
|
|
|
def test_fingerprint_rejects_dropped_current_truth_table_presence() -> None:
|
|
source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
|
|
restored = _with_legitimate_reconciliation_delta(
|
|
source, introduce_checkpoint=False
|
|
)
|
|
physical = restored["physical_tables"]
|
|
assert isinstance(physical, list)
|
|
physical.remove("host_telemetry_latest")
|
|
|
|
contract = fingerprint_compatibility(
|
|
source, restored, source_schema_revision="20260830_0024"
|
|
)
|
|
assert contract.compatible is False
|
|
|
|
|
|
def test_postgres_query_rows_parses_csv_safely_and_enforces_its_bound() -> None:
|
|
engine = PostgresEngine()
|
|
engine._run = lambda *_args, **_kwargs: CommandResult( # type: ignore[method-assign]
|
|
command="psql",
|
|
returncode=0,
|
|
stdout='id,details\r\n1,"{""message"":""a,b""}"\r\n',
|
|
stderr="",
|
|
duration_seconds=0.01,
|
|
)
|
|
assert engine.query_rows(TARGET, "select bounded", max_rows=1) == [
|
|
{"id": "1", "details": '{"message":"a,b"}'}
|
|
]
|
|
|
|
engine._run = lambda *_args, **_kwargs: CommandResult( # type: ignore[method-assign]
|
|
command="psql",
|
|
returncode=0,
|
|
stdout="id\r\n1\r\n2\r\n",
|
|
stderr="",
|
|
duration_seconds=0.01,
|
|
)
|
|
with pytest.raises(PostgresToolError, match="more than 1"):
|
|
engine.query_rows(TARGET, "select unexpectedly_unbounded", max_rows=1)
|