Implement hosted Conversion Attestation publication (WP-0006-T06)

migrations/0003_attestations.sql: one row per Phase, no UPDATE/DELETE
grant for trf_app (a published attestation is permanent).

src/target_revenue/attestation.py: publish_attestation() is idempotent
(existing row returned unchanged, never regenerated) and derives
conversion_timestamp from the ledger itself via _find_conversion_prefix(),
which finds the earliest prefix where the fold reaches Outstanding
Target = 0 rather than trusting entries[-1] or wall-clock publish time -
so a later unrelated entry can never change an already-converged Phase's
recorded conversion moment. Raises NotConvertedError rather than
fabricating a conversion. Reuses conversion.generate_attestation()
unchanged. Guarded the psycopg import behind TYPE_CHECKING so the pure
helper stays testable under plain system Python.

service/app.py adds GET /phases/{id}/attestation (unauthenticated,
publish-on-first-observation).

tests/test_attestation.py (3 tests, no Docker/Postgres) proves the
earliest-crossing behavior. 6 new Docker-gated tests in
test_ledger_hosting.py cover pre-conversion 404, the core
legal-technical property that /metrics facts/calculations/forecasts
are identical whether or not /attestation was ever called, one-time
publication, signature verification, and DB-level UPDATE/DELETE
rejection.
This commit is contained in:
tegwick 2026-07-29 21:50:02 +02:00
parent caa98ead3b
commit 28f0f429b2
7 changed files with 341 additions and 3 deletions

View file

@ -30,6 +30,7 @@ REPO_ROOT = Path(__file__).resolve().parents[1]
MIGRATIONS = [
REPO_ROOT / "migrations" / "0001_registries.sql",
REPO_ROOT / "migrations" / "0002_ledger.sql",
REPO_ROOT / "migrations" / "0003_attestations.sql",
]
pytestmark = pytest.mark.skipif(
@ -301,3 +302,103 @@ def test_metrics_endpoint_unauthenticated_and_matches_offline_computation(
assert hosted == offline
assert hosted["facts"]["cumulative_development_credit"] == 4000
def test_attestation_not_available_before_conversion(client, pg_container, registered_phase):
phase_id = registered_phase["phase"]["id"]
e = _entry(phase_id, "trsl:entry:ledgertestattest0001", "development-credit", 1, "2026-08-01T00:00:00Z")
resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
assert resp.status_code == 201
metrics_resp = client.get(f"/phases/{phase_id}/metrics")
assert metrics_resp.json()["facts"]["is_converted"] is False
attestation_resp = client.get(f"/phases/{phase_id}/attestation")
assert attestation_resp.status_code == 404
def test_conversion_is_true_independent_of_attestation_call(client, pg_container):
"""TSD §3.5 legal-technical rule: conversion is already true from the
ledger the instant the fold reaches zero; nothing about calling (or not
calling) /attestation may change what /metrics already reports."""
manifest = golden_manifest()
manifest["phase"]["id"] = manifest["phase"]["id"] + "-attest-conv-" + uuid.uuid4().hex[:6]
manifest["phase"]["initial_target"]["amount"] = 1000
client.post("/phases", json=manifest, headers=auth_headers(pg_container["token"])).raise_for_status()
phase_id = manifest["phase"]["id"]
e = _entry(phase_id, "trsl:entry:ledgertestattest0002", "development-credit", 1000, "2026-08-01T00:00:00Z")
client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"])).raise_for_status()
before = client.get(f"/phases/{phase_id}/metrics").json()
assert before["facts"]["is_converted"] is True
assert before["facts"]["outstanding_target"] == 0
# No attestation has been requested yet; conversion is already true.
attestation_resp = client.get(f"/phases/{phase_id}/attestation")
assert attestation_resp.status_code == 200
after = client.get(f"/phases/{phase_id}/metrics").json()
assert after["facts"]["is_converted"] is True
# Publishing the attestation must change nothing about the conversion
# facts/calculations/forecasts themselves (only `as_of` legitimately
# differs, since /metrics stamps wall-clock time on every call).
assert after["facts"] == before["facts"]
assert after["calculations"] == before["calculations"]
assert after["forecasts"] == before["forecasts"]
def test_attestation_published_once_and_idempotent_on_reread(client, pg_container):
manifest = golden_manifest()
manifest["phase"]["id"] = manifest["phase"]["id"] + "-attest-idem-" + uuid.uuid4().hex[:6]
manifest["phase"]["initial_target"]["amount"] = 500
client.post("/phases", json=manifest, headers=auth_headers(pg_container["token"])).raise_for_status()
phase_id = manifest["phase"]["id"]
e = _entry(phase_id, "trsl:entry:ledgertestattest0003", "development-credit", 500, "2026-08-01T00:00:00Z")
client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"])).raise_for_status()
first = client.get(f"/phases/{phase_id}/attestation")
assert first.status_code == 200
second = client.get(f"/phases/{phase_id}/attestation")
assert second.status_code == 200
assert first.json() == second.json() # same signed record, not regenerated
from target_revenue import validation
validation.validate_conversion_attestation(
{k: v for k, v in first.json().items() if k != "signature"}
)
assert first.json()["conversion_timestamp"] == "2026-08-01T00:00:00Z"
assert first.json()["ledger_checkpoint"] == "trsl:entry:ledgertestattest0003"
def test_attestation_verifiable_with_public_key(client, pg_container):
manifest = golden_manifest()
manifest["phase"]["id"] = manifest["phase"]["id"] + "-attest-verify-" + uuid.uuid4().hex[:6]
manifest["phase"]["initial_target"]["amount"] = 500
client.post("/phases", json=manifest, headers=auth_headers(pg_container["token"])).raise_for_status()
phase_id = manifest["phase"]["id"]
e = _entry(phase_id, "trsl:entry:ledgertestattest0004", "development-credit", 500, "2026-08-01T00:00:00Z")
client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"])).raise_for_status()
published = client.get(f"/phases/{phase_id}/attestation").json()
pk_hex = client.get("/public-key").json()["public_key_hex"]
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from target_revenue import hashing
public_key = Ed25519PublicKey.from_public_bytes(bytes.fromhex(pk_hex))
assert hashing.verify_record_signature(published, published["signature"], public_key)
def test_application_role_cannot_update_or_delete_attestations(pg_container):
with psycopg.connect(pg_container["app_dsn"]) as conn:
with pytest.raises(psycopg.errors.InsufficientPrivilege):
conn.execute("UPDATE attestations SET signature = 'x' WHERE phase_id = 'nonexistent'")
conn.rollback()
with pytest.raises(psycopg.errors.InsufficientPrivilege):
conn.execute("DELETE FROM attestations WHERE phase_id = 'nonexistent'")
conn.rollback()