tegwick
885da0a1cb
migrations/0006_control_plane.sql: control_plane_audit_log (append-only, no UPDATE/DELETE for trf_app) and control_plane_proposed_entries (the Contributor tier's "propose, don't append" workflow from concept §2) - review decisions go through a review_proposed_entry() SECURITY DEFINER function, same governance-action pattern as set_extension_status/revoke_credential, not a direct UPDATE. src/target_revenue/control_plane.py is the enforcement layer concept §2 called for: register_phase/append_development_credit require Operator+; propose_ledger_entry requires Contributor+ and stores a pending proposal without touching the real Ledger; approve_proposed_entry (Operator+) appends it under the *reviewer's own* credential/attribution (not the original proposer's - the reviewer is who's authorizing it into the real Ledger, while the proposer stays on record in the proposal row and audit log); reject_proposed_entry (Operator+) discards it. issue_/ revoke_user_credential (Admin+) wrap registry.py's T02 functions with the same rights check and audit logging. Every action funnels through record_audit_event, independent of the Trust Service's own signed records. tests/test_control_plane.py (12 tests): rights enforcement at each tier boundary, the full propose -> approve -> appended-under-reviewer flow, propose -> reject -> nothing appended, double-review rejection, audit log content/attribution, DB-level UPDATE rejection on both new tables. Full suite: 84 offline (unchanged), 53 with Docker (up from 41); no stray containers left running.
343 lines
12 KiB
Python
343 lines
12 KiB
Python
"""Integration tests for WP-0009-T03 (Control Plane backend: rights
|
|
enforcement + audit log + proposed-entry review).
|
|
|
|
Same ephemeral, disposable Postgres-via-Docker pattern as the other
|
|
hosted test modules (never the shared state-hub instance).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import subprocess
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
psycopg = pytest.importorskip("psycopg")
|
|
|
|
from conftest import golden_manifest # noqa: E402
|
|
|
|
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",
|
|
REPO_ROOT / "migrations" / "0004_breach_records.sql",
|
|
REPO_ROOT / "migrations" / "0005_licensor_credentials.sql",
|
|
REPO_ROOT / "migrations" / "0006_control_plane.sql",
|
|
]
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
shutil.which("docker") is None, reason="docker not available"
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def pg_container():
|
|
name = f"trf-test-pg-cp-{uuid.uuid4().hex[:8]}"
|
|
subprocess.run(
|
|
[
|
|
"docker", "run", "--rm", "-d",
|
|
"--name", name,
|
|
"-e", "POSTGRES_PASSWORD=postgres",
|
|
"-e", "POSTGRES_DB=target_revenue_test",
|
|
"-p", "127.0.0.1::5432",
|
|
"postgres:16-alpine",
|
|
],
|
|
check=True, capture_output=True,
|
|
)
|
|
try:
|
|
port_out = subprocess.run(
|
|
["docker", "port", name, "5432/tcp"], check=True, capture_output=True, text=True
|
|
).stdout.strip()
|
|
host_port = port_out.split(":")[-1]
|
|
dsn = f"host=127.0.0.1 port={host_port} dbname=target_revenue_test user=postgres password=postgres"
|
|
|
|
for _ in range(60):
|
|
try:
|
|
with psycopg.connect(dsn, connect_timeout=1):
|
|
break
|
|
except psycopg.OperationalError:
|
|
time.sleep(0.5)
|
|
else:
|
|
raise RuntimeError("postgres container did not become ready in time")
|
|
|
|
with psycopg.connect(dsn) as admin_conn:
|
|
for migration in MIGRATIONS:
|
|
admin_conn.execute(migration.read_text(encoding="utf-8"))
|
|
admin_conn.commit()
|
|
admin_conn.execute(
|
|
"INSERT INTO licensors (token, licensor_id, credential_label, rights, issued_by) "
|
|
"VALUES (%s, %s, %s, %s, %s)",
|
|
("founding-admin-token", "binky", "founder", "admin", "bootstrap"),
|
|
)
|
|
admin_conn.commit()
|
|
|
|
app_dsn = (
|
|
f"host=127.0.0.1 port={host_port} dbname=target_revenue_test "
|
|
f"user=trf_app password=changeme-in-deployment"
|
|
)
|
|
yield {"admin_dsn": dsn, "app_dsn": app_dsn, "admin_token": "founding-admin-token"}
|
|
finally:
|
|
subprocess.run(["docker", "stop", name], capture_output=True)
|
|
|
|
|
|
@pytest.fixture()
|
|
def conn(pg_container):
|
|
with psycopg.connect(pg_container["app_dsn"]) as connection:
|
|
yield connection
|
|
|
|
|
|
@pytest.fixture()
|
|
def signing_key():
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
|
|
return Ed25519PrivateKey.generate()
|
|
|
|
|
|
def _entry(phase_id: str, entry_id: str) -> dict:
|
|
return {
|
|
"id": entry_id,
|
|
"phase": phase_id,
|
|
"type": "development-credit",
|
|
"amount": 100,
|
|
"currency": "USD",
|
|
"recognized_at": "2026-08-01T00:00:00Z",
|
|
"evidence_reference": f"confidential:evidence:{entry_id}",
|
|
"extension": {"id": "trsl:extension:development-license", "version": "1.0"},
|
|
}
|
|
|
|
|
|
@pytest.fixture()
|
|
def credentials(conn, pg_container):
|
|
"""One credential per rights tier, all under the `binky` tenant.
|
|
Labels are uniquified per test invocation — `pg_container` is
|
|
module-scoped (one Postgres for the whole file), so reusing fixed
|
|
labels across tests would collide with the active-label uniqueness
|
|
constraint from a previous test's still-active credential."""
|
|
from target_revenue import registry
|
|
|
|
admin = registry.authenticate(conn, pg_container["admin_token"])
|
|
suffix = uuid.uuid4().hex[:8]
|
|
tiers = {}
|
|
for label, rights in [
|
|
("viewer-user", "viewer"),
|
|
("contributor-user", "contributor"),
|
|
("operator-user", "operator"),
|
|
]:
|
|
cred = registry.issue_sub_credential(
|
|
conn, licensor_id="binky", credential_label=f"{label}-{suffix}", rights=rights,
|
|
issued_by="founder",
|
|
)
|
|
tiers[rights] = cred
|
|
conn.commit()
|
|
tiers["admin"] = admin
|
|
return tiers
|
|
|
|
|
|
@pytest.fixture()
|
|
def registered_phase(conn, credentials):
|
|
from target_revenue import control_plane
|
|
|
|
manifest = golden_manifest()
|
|
manifest["phase"]["id"] = manifest["phase"]["id"] + "-cp-test-" + uuid.uuid4().hex[:6]
|
|
control_plane.register_phase(conn, credentials["operator"], manifest)
|
|
conn.commit()
|
|
return manifest
|
|
|
|
|
|
# --- rights enforcement -----------------------------------------------------
|
|
|
|
|
|
def test_viewer_cannot_register_phase(conn, credentials):
|
|
from target_revenue import control_plane
|
|
|
|
manifest = golden_manifest()
|
|
manifest["phase"]["id"] = manifest["phase"]["id"] + "-viewer-" + uuid.uuid4().hex[:6]
|
|
with pytest.raises(control_plane.ControlPlaneError, match="insufficient rights"):
|
|
control_plane.register_phase(conn, credentials["viewer"], manifest)
|
|
|
|
|
|
def test_contributor_cannot_append_directly(conn, credentials, registered_phase, signing_key):
|
|
from target_revenue import control_plane
|
|
|
|
phase_id = registered_phase["phase"]["id"]
|
|
with pytest.raises(control_plane.ControlPlaneError, match="insufficient rights"):
|
|
control_plane.append_development_credit(
|
|
conn, credentials["contributor"], phase_id, _entry(phase_id, "trsl:entry:cpdirect0001"), signing_key
|
|
)
|
|
|
|
|
|
def test_operator_can_register_and_append(conn, credentials, signing_key):
|
|
from target_revenue import control_plane
|
|
|
|
manifest = golden_manifest()
|
|
manifest["phase"]["id"] = manifest["phase"]["id"] + "-opappend-" + uuid.uuid4().hex[:6]
|
|
control_plane.register_phase(conn, credentials["operator"], manifest)
|
|
conn.commit()
|
|
|
|
stored = control_plane.append_development_credit(
|
|
conn, credentials["operator"], manifest["phase"]["id"],
|
|
_entry(manifest["phase"]["id"], "trsl:entry:cpop0001"), signing_key,
|
|
)
|
|
conn.commit()
|
|
assert stored["id"] == "trsl:entry:cpop0001"
|
|
|
|
|
|
# --- proposed-entry review workflow -----------------------------------------
|
|
|
|
|
|
def test_contributor_can_propose_but_not_review(conn, credentials, registered_phase):
|
|
from target_revenue import control_plane
|
|
|
|
phase_id = registered_phase["phase"]["id"]
|
|
proposal_id = control_plane.propose_ledger_entry(
|
|
conn, credentials["contributor"], phase_id, _entry(phase_id, "trsl:entry:cppropose0001")
|
|
)
|
|
conn.commit()
|
|
assert isinstance(proposal_id, int)
|
|
|
|
pending = control_plane.list_proposed_entries(conn, phase_id=phase_id, status="pending")
|
|
assert len(pending) == 1
|
|
assert pending[0]["id"] == proposal_id
|
|
|
|
with pytest.raises(control_plane.ControlPlaneError, match="insufficient rights"):
|
|
control_plane.approve_proposed_entry(
|
|
conn, credentials["contributor"], proposal_id, signing_key=None
|
|
)
|
|
|
|
|
|
def test_operator_approves_proposal_appends_under_own_credential(
|
|
conn, credentials, registered_phase, signing_key
|
|
):
|
|
from target_revenue import control_plane
|
|
|
|
phase_id = registered_phase["phase"]["id"]
|
|
proposal_id = control_plane.propose_ledger_entry(
|
|
conn, credentials["contributor"], phase_id, _entry(phase_id, "trsl:entry:cpapprove0001")
|
|
)
|
|
conn.commit()
|
|
|
|
stored = control_plane.approve_proposed_entry(
|
|
conn, credentials["operator"], proposal_id, signing_key, review_note="looks right"
|
|
)
|
|
conn.commit()
|
|
assert stored["id"] == "trsl:entry:cpapprove0001"
|
|
|
|
remaining_pending = control_plane.list_proposed_entries(conn, phase_id=phase_id, status="pending")
|
|
assert remaining_pending == []
|
|
approved = control_plane.list_proposed_entries(conn, phase_id=phase_id, status="approved")
|
|
assert len(approved) == 1
|
|
|
|
# Attribution: proposer and approver are both independently visible.
|
|
from target_revenue import ledger
|
|
|
|
attribution = ledger.get_ledger_attribution(conn, phase_id)
|
|
assert attribution[0]["credential_label"] == credentials["operator"].credential_label
|
|
|
|
|
|
def test_operator_rejects_proposal_nothing_appended(conn, credentials, registered_phase):
|
|
from target_revenue import control_plane, ledger
|
|
|
|
phase_id = registered_phase["phase"]["id"]
|
|
proposal_id = control_plane.propose_ledger_entry(
|
|
conn, credentials["contributor"], phase_id, _entry(phase_id, "trsl:entry:cpreject0001")
|
|
)
|
|
conn.commit()
|
|
|
|
control_plane.reject_proposed_entry(conn, credentials["operator"], proposal_id, review_note="not this one")
|
|
conn.commit()
|
|
|
|
rejected = control_plane.list_proposed_entries(conn, phase_id=phase_id, status="rejected")
|
|
assert len(rejected) == 1
|
|
assert ledger.get_ledger(conn, phase_id) == []
|
|
|
|
|
|
def test_double_review_of_same_proposal_rejected(conn, credentials, registered_phase, signing_key):
|
|
from target_revenue import control_plane
|
|
|
|
phase_id = registered_phase["phase"]["id"]
|
|
proposal_id = control_plane.propose_ledger_entry(
|
|
conn, credentials["contributor"], phase_id, _entry(phase_id, "trsl:entry:cpdouble0001")
|
|
)
|
|
conn.commit()
|
|
|
|
control_plane.approve_proposed_entry(conn, credentials["operator"], proposal_id, signing_key)
|
|
conn.commit()
|
|
|
|
with pytest.raises(control_plane.ControlPlaneError, match="no pending proposal"):
|
|
control_plane.reject_proposed_entry(conn, credentials["operator"], proposal_id)
|
|
|
|
|
|
# --- audit log ---------------------------------------------------------------
|
|
|
|
|
|
def test_audit_log_records_actor_and_action(conn, credentials, registered_phase, signing_key):
|
|
from target_revenue import control_plane
|
|
|
|
phase_id = registered_phase["phase"]["id"]
|
|
control_plane.append_development_credit(
|
|
conn, credentials["operator"], phase_id, _entry(phase_id, "trsl:entry:cpaudit0001"), signing_key
|
|
)
|
|
conn.commit()
|
|
|
|
log = control_plane.get_audit_log(conn, phase_id=phase_id)
|
|
actions = [row["action"] for row in log]
|
|
assert "register_phase" in actions
|
|
assert "append_ledger_entry" in actions
|
|
append_row = next(row for row in log if row["action"] == "append_ledger_entry")
|
|
assert append_row["actor_credential_label"] == credentials["operator"].credential_label
|
|
assert append_row["trust_service_record_id"] == "trsl:entry:cpaudit0001"
|
|
|
|
|
|
# --- admin credential management ---------------------------------------------
|
|
|
|
|
|
def test_admin_can_issue_and_revoke_credential(conn, credentials):
|
|
from target_revenue import control_plane, registry
|
|
|
|
new_cred = control_plane.issue_user_credential(
|
|
conn, credentials["admin"], licensor_id="binky", credential_label="dana", rights="viewer"
|
|
)
|
|
conn.commit()
|
|
assert registry.authenticate(conn, new_cred.token).credential_label == "dana"
|
|
|
|
control_plane.revoke_user_credential(conn, credentials["admin"], new_cred.token)
|
|
conn.commit()
|
|
with pytest.raises(registry.RegistrationError):
|
|
registry.authenticate(conn, new_cred.token)
|
|
|
|
|
|
def test_non_admin_cannot_issue_credential(conn, credentials):
|
|
from target_revenue import control_plane
|
|
|
|
with pytest.raises(control_plane.ControlPlaneError, match="insufficient rights"):
|
|
control_plane.issue_user_credential(
|
|
conn, credentials["operator"], licensor_id="binky", credential_label="eve", rights="viewer"
|
|
)
|
|
|
|
|
|
# --- DB-level append-only enforcement ----------------------------------------
|
|
|
|
|
|
def test_application_role_cannot_update_audit_log(pg_container):
|
|
with psycopg.connect(pg_container["app_dsn"]) as app_conn:
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
app_conn.execute(
|
|
"UPDATE control_plane_audit_log SET action = 'tampered' WHERE id = 1"
|
|
)
|
|
app_conn.rollback()
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
app_conn.execute("DELETE FROM control_plane_audit_log WHERE id = 1")
|
|
app_conn.rollback()
|
|
|
|
|
|
def test_application_role_cannot_update_proposed_entries_directly(pg_container):
|
|
with psycopg.connect(pg_container["app_dsn"]) as app_conn:
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
app_conn.execute(
|
|
"UPDATE control_plane_proposed_entries SET status = 'approved' WHERE id = 1"
|
|
)
|
|
app_conn.rollback()
|