coulomb/target-revenue
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement Control Plane backend: rights enforcement + audit log (WP-0009-T03)

Browse source 
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.
This commit is contained in:
tegwick 2026-07-30 15:08:59 +02:00
parent d29447fcff
commit 885da0a1cb
6 changed files with 756 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

2
README.md
View file

@ -81,7 +81,7 @@ The concept's §13 now defines a **Global Contingency Share Determination Rule**
| [TREV-WP-0006](workplans/TREV-WP-0006-trust-service-implementation.md) | Hosted Trust Service reference implementation (PRD Phase 4b) — **finished**, all 9 tasks done (Postgres-backed registries/ledger/metrics/attestation/breach-record, ADR-0002 accepted, onboarding CLI, hosted conformance suite) |
| [TREV-WP-0007](workplans/TREV-WP-0007-degeneration-policy-and-canonical-profiles.md) | Degeneration policy + canonical monetization profile catalog — **finished**, all 4 tasks done. `trsl:policy:linear-longstop-v0` confirmed 2026-07-29 as the v1 norm for the first pilot cohort; `progress-paused-longstop-v1` named as the next iteration, not yet adopted |
| [TREV-WP-0008](workplans/TREV-WP-0008-governance-and-pilot-rollout.md) | Governance formalization + pilot rollout — active; T01–T04 done. `info-tech-canon` dry-run onboarding routine exercised end-to-end 2026-07-29. **Org-wide TRSL license adoption executed 2026-07-30** across ~90 `coulomb`-org repos (`history/260730-TRSL-OrgWideLicenseRollout.md`) — a license-text adoption, not a Phase declaration. T05 (real Phase go-live gate) remains `todo` by design; no Phase exists yet for any repo |
| [TREV-WP-0009](workplans/TREV-WP-0009-target-revenue-control-plane.md) | Target Revenue Control Plane — interactive UI for the `binky` tenant, incl. interactive Development Credit entry creation (`specs/TargetRevenueControlPlaneConcept.md`) — active; T01 (rights model) and **T02 (WP-0006 auth extension for per-human sub-credentials, `migrations/0005_licensor_credentials.sql`) done**; T03 (Control Plane backend) next |
| [TREV-WP-0009](workplans/TREV-WP-0009-target-revenue-control-plane.md) | Target Revenue Control Plane — interactive UI for the `binky` tenant, incl. interactive Development Credit entry creation (`specs/TargetRevenueControlPlaneConcept.md`) — active; T01, T02, and **T03 (backend rights enforcement + audit log + propose/review workflow, `src/target_revenue/control_plane.py`) done**; T04 (interactive UI) next |
| [TREV-WP-0010](workplans/TREV-WP-0010-development-effort-calculator.md) | Development Effort Calculator — **finished**, all 3 tasks done. Applied to the three real pilot candidates (`history/260730-EffortCalculator-CandidateApplication.md`) — every calculator-derived Initial Target came out materially lower than the earlier hand-picked placeholders, two of three carrying explicit warnings recommending manual review |
Hub index: [`WORK-RECORDS.md`](WORK-RECORDS.md) · brief: [`.custodian-brief.md`](.custodian-brief.md)

92
migrations/0006_control_plane.sql Normal file
View file

@ -0,0 +1,92 @@
-- WP-0009-T03: Control Plane backend — audit log and proposed-entry review.
-- Depends on migrations/0001_registries.sql, 0002_ledger.sql, and
-- 0005_licensor_credentials.sql (credential_label/rights/licensor_identities).
--
-- These tables belong to the Control Plane, a client layer on top of the
-- already-finished hosted Trust Service (registry.py/ledger.py) — not new
-- Trust Service surface itself. Per
-- specs/TargetRevenueControlPlaneConcept.md §5: the audit log here is
-- deliberately separate from the Trust Service's own signed records
-- (phase_manifests, ledger_entries, attestations, breach_records), which
-- only ever attest "the tenant did this," never the individual human.
BEGIN;
-- Append-only audit trail: which credential (hence which human) took
-- which Control Plane action, and — where applicable — which Trust
-- Service record resulted. `trf_app` has no UPDATE/DELETE grant: an audit
-- log that could be edited after the fact isn't an audit log.
CREATE TABLE IF NOT EXISTS control_plane_audit_log (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
actor_token text NOT NULL REFERENCES licensors(token),
actor_licensor_id text NOT NULL,
actor_credential_label text,
action text NOT NULL,
phase_id text,
trust_service_record_id text,
detail jsonb,
created_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS control_plane_audit_log_phase_idx
ON control_plane_audit_log (phase_id);
GRANT SELECT, INSERT ON control_plane_audit_log TO trf_app;
-- Contributor-tier proposed entries, per
-- specs/TargetRevenueControlPlaneConcept.md §2's rights table: a
-- Contributor may submit a proposed Development Credit entry but not
-- append it directly; an Operator/Admin reviews and either approves
-- (which actually appends it to the real Ledger under the reviewer's own
-- credential) or rejects it. `trf_app` again has no UPDATE/DELETE grant —
-- review decisions are recorded via a SECURITY DEFINER function, the same
-- governance-action pattern as `set_extension_status`/`revoke_credential`.
CREATE TABLE IF NOT EXISTS control_plane_proposed_entries (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
phase_id text NOT NULL,
entry jsonb NOT NULL,
proposed_by_token text NOT NULL REFERENCES licensors(token),
status text NOT NULL DEFAULT 'pending'
CHECK (status IN ('pending', 'approved', 'rejected')),
reviewed_by_token text REFERENCES licensors(token),
reviewed_at timestamptz,
review_note text,
appended_entry_id text,
created_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS control_plane_proposed_entries_phase_idx
ON control_plane_proposed_entries (phase_id, status);
GRANT SELECT, INSERT ON control_plane_proposed_entries TO trf_app;
CREATE OR REPLACE FUNCTION review_proposed_entry(
p_id bigint,
p_status text,
p_reviewed_by_token text,
p_review_note text,
p_appended_entry_id text
) RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
BEGIN
IF p_status NOT IN ('approved', 'rejected') THEN
RAISE EXCEPTION 'review_proposed_entry only permits approved/rejected, got %', p_status;
END IF;
UPDATE control_plane_proposed_entries
SET status = p_status,
reviewed_by_token = p_reviewed_by_token,
reviewed_at = now(),
review_note = p_review_note,
appended_entry_id = p_appended_entry_id
WHERE id = p_id AND status = 'pending';
IF NOT FOUND THEN
RAISE EXCEPTION 'no pending proposal with id %', p_id;
END IF;
END;
$$;
GRANT EXECUTE ON FUNCTION review_proposed_entry(bigint, text, text, text, text) TO trf_app;
COMMIT;

9
specs/TargetRevenueControlPlaneConcept.md
View file

@ -122,6 +122,15 @@ Licensor token (§2). This log is the answer to "who actually clicked the
button," which matters operationally even though it's not part of the
Trust Service's own cryptographic guarantees.
**Implemented 2026-07-30** (`workplans/TREV-WP-0009-target-revenue-control-plane.md`
T03, `src/target_revenue/control_plane.py`): `control_plane_audit_log`
(append-only) records every action; the Contributor tier's "propose, not
append" workflow (§2's rights table) is backed by
`control_plane_proposed_entries`, with Operator/Admin review going
through a governance-gated database function rather than a plain
UPDATE — the same pattern already established for extension
canonicalization and credential revocation.
## 6. Explicit non-goals
- Replacing `scripts/trf_onboard.py` — the CLI remains valid for

279
src/target_revenue/control_plane.py Normal file
View file

@ -0,0 +1,279 @@
"""Target Revenue Control Plane backend (WP-0009-T03).
A client layer over the already-finished hosted Trust Service
(`registry.py`, `ledger.py`) — not a replacement for it, per
`specs/TargetRevenueControlPlaneConcept.md`. Every write-capable Control
Plane action funnels through this module so it can (a) check the acting
credential's rights tier before doing anything, and (b) record who did
what in `control_plane_audit_log`, independent of the Trust Service's own
signed records, which only ever attest "the tenant did this," never the
individual human (concept §2; `migrations/0005_licensor_credentials.sql`).
Rights enforcement lives here, not in `registry.py`/`ledger.py` — those
modules only ever needed to answer "which tenant is this," and per
WP-0009-T01's accepted decision, per-human rights tiers
(Viewer/Contributor/Operator/Admin) are a Control Plane concern layered on
top, not a Trust Service one.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from . import ledger, registry
from .registry import Licensor
if TYPE_CHECKING:
from psycopg import Connection
class ControlPlaneError(ValueError):
"""Raised when a Control Plane action is rejected (insufficient
rights, an unknown proposal id, etc.)."""
def _require_right(licensor: Licensor, minimum: str) -> None:
if not registry.has_right(licensor.rights, minimum):
raise ControlPlaneError(
f"insufficient rights: {minimum!r} required, this credential has "
f"{licensor.rights!r}"
)
def record_audit_event(
conn: "Connection",
actor: Licensor,
action: str,
phase_id: str | None = None,
trust_service_record_id: str | None = None,
detail: dict[str, Any] | None = None,
) -> None:
"""Append one row to the Control Plane's own audit log. Called by
every function below after (never instead of, and never before
confirming) the underlying Trust Service action actually succeeded."""
from psycopg.types.json import Jsonb
conn.execute(
"""
INSERT INTO control_plane_audit_log
(actor_token, actor_licensor_id, actor_credential_label, action,
phase_id, trust_service_record_id, detail)
VALUES (%s, %s, %s, %s, %s, %s, %s)
""",
(
actor.token,
actor.licensor_id,
actor.credential_label,
action,
phase_id,
trust_service_record_id,
Jsonb(detail) if detail is not None else None,
),
)
def get_audit_log(conn: "Connection", phase_id: str | None = None) -> list[dict[str, Any]]:
"""Read the audit log, optionally scoped to one Phase, in append order."""
if phase_id:
rows = conn.execute(
"""
SELECT id, actor_licensor_id, actor_credential_label, action, phase_id,
trust_service_record_id, detail, created_at
FROM control_plane_audit_log
WHERE phase_id = %s
ORDER BY id ASC
""",
(phase_id,),
).fetchall()
else:
rows = conn.execute(
"""
SELECT id, actor_licensor_id, actor_credential_label, action, phase_id,
trust_service_record_id, detail, created_at
FROM control_plane_audit_log
ORDER BY id ASC
"""
).fetchall()
columns = (
"id", "actor_licensor_id", "actor_credential_label", "action", "phase_id",
"trust_service_record_id", "detail", "created_at",
)
return [dict(zip(columns, row)) for row in rows]
# --- Human-user (credential) management: Admin tier -------------------------
def issue_user_credential(
conn: "Connection", admin: Licensor, licensor_id: str, credential_label: str, rights: str
) -> Licensor:
"""Admin-only: issue a new credential for a human user acting as the
given Licensor tenant. Wraps `registry.issue_sub_credential` with the
rights check and audit log this module exists to add."""
_require_right(admin, "admin")
new_credential = registry.issue_sub_credential(
conn,
licensor_id=licensor_id,
credential_label=credential_label,
rights=rights,
issued_by=admin.credential_label or admin.token,
)
record_audit_event(
conn, admin, action="issue_user_credential",
detail={"credential_label": credential_label, "rights": rights},
)
return new_credential
def revoke_user_credential(conn: "Connection", admin: Licensor, token: str) -> None:
"""Admin-only: revoke a credential. Logs only the revoked token's
suffix, not the full secret, in the audit detail."""
_require_right(admin, "admin")
registry.revoke_sub_credential(conn, token)
record_audit_event(
conn, admin, action="revoke_user_credential",
detail={"revoked_token_suffix": token[-6:]},
)
# --- Phase / Ledger actions --------------------------------------------------
def register_phase(conn: "Connection", licensor: Licensor, manifest: dict[str, Any]) -> None:
"""Operator+ only, per concept §2's rights table."""
_require_right(licensor, "operator")
registry.register_phase_manifest(conn, licensor, manifest)
record_audit_event(
conn, licensor, action="register_phase", phase_id=manifest["phase"]["id"],
detail={"milestone_release": manifest["phase"]["milestone_release"]["name"]},
)
def append_development_credit(
conn: "Connection",
licensor: Licensor,
phase_id: str,
entry_input: dict[str, Any],
signing_key: Any,
) -> dict[str, Any]:
"""Operator+ only — appends directly. A Contributor must use
`propose_ledger_entry` instead (concept §2: "not directly appended")."""
_require_right(licensor, "operator")
stored = ledger.append_entry(conn, licensor, phase_id, entry_input, signing_key)
record_audit_event(
conn, licensor, action="append_ledger_entry", phase_id=phase_id,
trust_service_record_id=stored["id"],
detail={"type": stored["type"], "amount": stored["amount"]},
)
return stored
def propose_ledger_entry(
conn: "Connection", licensor: Licensor, phase_id: str, entry_input: dict[str, Any]
) -> int:
"""Contributor+ — submit a proposed entry for an Operator/Admin to
review. Not appended to the real Ledger until approved."""
from psycopg.types.json import Jsonb
_require_right(licensor, "contributor")
row = conn.execute(
"""
INSERT INTO control_plane_proposed_entries (phase_id, entry, proposed_by_token)
VALUES (%s, %s, %s)
RETURNING id
""",
(phase_id, Jsonb(entry_input), licensor.token),
).fetchone()
proposal_id = row[0]
record_audit_event(
conn, licensor, action="propose_ledger_entry", phase_id=phase_id,
detail={"proposal_id": proposal_id, "type": entry_input.get("type")},
)
return proposal_id
def list_proposed_entries(
conn: "Connection", phase_id: str | None = None, status: str = "pending"
) -> list[dict[str, Any]]:
if phase_id:
rows = conn.execute(
"""
SELECT id, phase_id, entry, proposed_by_token, status, created_at
FROM control_plane_proposed_entries
WHERE phase_id = %s AND status = %s
ORDER BY id ASC
""",
(phase_id, status),
).fetchall()
else:
rows = conn.execute(
"""
SELECT id, phase_id, entry, proposed_by_token, status, created_at
FROM control_plane_proposed_entries
WHERE status = %s
ORDER BY id ASC
""",
(status,),
).fetchall()
columns = ("id", "phase_id", "entry", "proposed_by_token", "status", "created_at")
return [dict(zip(columns, row)) for row in rows]
def approve_proposed_entry(
conn: "Connection",
reviewer: Licensor,
proposal_id: int,
signing_key: Any,
review_note: str | None = None,
) -> dict[str, Any]:
"""Operator+ approves: actually appends the entry, under the
*reviewer's* own credential/attribution — the reviewer is the one
authorizing it into the real Ledger. The original proposer remains on
record in the proposal row and the audit log, not lost or overwritten.
"""
_require_right(reviewer, "operator")
row = conn.execute(
"SELECT phase_id, entry FROM control_plane_proposed_entries "
"WHERE id = %s AND status = 'pending'",
(proposal_id,),
).fetchone()
if row is None:
raise ControlPlaneError(f"no pending proposal with id {proposal_id!r}")
phase_id, entry_input = row
stored = ledger.append_entry(conn, reviewer, phase_id, entry_input, signing_key)
conn.execute(
"SELECT review_proposed_entry(%s, %s, %s, %s, %s)",
(proposal_id, "approved", reviewer.token, review_note, stored["id"]),
)
record_audit_event(
conn, reviewer, action="approve_proposed_entry", phase_id=phase_id,
trust_service_record_id=stored["id"], detail={"proposal_id": proposal_id},
)
return stored
def reject_proposed_entry(
conn: "Connection", reviewer: Licensor, proposal_id: int, review_note: str | None = None
) -> None:
"""Operator+ rejects: the proposal is marked rejected, nothing is
appended to the real Ledger."""
_require_right(reviewer, "operator")
row = conn.execute(
"SELECT phase_id FROM control_plane_proposed_entries "
"WHERE id = %s AND status = 'pending'",
(proposal_id,),
).fetchone()
if row is None:
raise ControlPlaneError(f"no pending proposal with id {proposal_id!r}")
phase_id = row[0]
conn.execute(
"SELECT review_proposed_entry(%s, %s, %s, %s, %s)",
(proposal_id, "rejected", reviewer.token, review_note, None),
)
record_audit_event(
conn, reviewer, action="reject_proposed_entry", phase_id=phase_id,
detail={"proposal_id": proposal_id},
)

343
tests/test_control_plane.py Normal file
View file

@ -0,0 +1,343 @@
"""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()

33
workplans/TREV-WP-0009-target-revenue-control-plane.md
View file

@ -143,7 +143,7 @@ suite: 84 passing offline (unchanged), 41 passing with Docker (up from
```task
id: TREV-WP-0009-T03
status: todo
status: done
priority: high
state_hub_task_id: "86a58e61-dccd-4678-b694-22a9eaea2b3c"
```
@ -156,6 +156,37 @@ the Trust Service's own signed record id for that action. This is the
piece that must exist before any write-capable UI flow (T04) can be built
responsibly.
**Result:** `migrations/0006_control_plane.sql` adds
`control_plane_audit_log` (append-only, no UPDATE/DELETE grant 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 rather than touching
the real Ledger; `approve_proposed_entry` (Operator+) appends it — under
the *reviewer's own* credential/attribution, not the original proposer's,
since the reviewer is the one 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 with nothing appended.
`issue_user_credential`/`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` — actor
credential label, action, Phase, and (where applicable) the resulting
Trust Service record id, independent of the Trust Service's own signed
records.
`tests/test_control_plane.py` (12 tests): rights enforcement at each
tier boundary (Viewer can't register, Contributor can't append directly
or review, non-Admin can't issue credentials), the full propose → approve
→ appended-under-reviewer's-credential flow, propose → reject → nothing
appended, double-review rejection, audit log content and attribution, and
DB-level UPDATE rejection on both new tables. Full suite: 84 passing
offline (unchanged), 53 passing with Docker (up from 41); no stray
containers left running.
## Control Plane interactive UI: Phase registration and Development Credit entry
```task