tegwick
7986e62f31
migrations/0005_licensor_credentials.sql: licensors can now hold multiple rows per licensor_id (credential_label, rights tier, issued_by, revoked_at). Real structural finding: licensor_id couldn't simply become non-unique, since phase_manifests, extensions, and breach_records all FK to licensors(licensor_id), which only worked because that column used to be unique. Introduced licensor_identities (one row per tenant) as the new FK target for all four tables, with an ensure_licensor_identity trigger auto-creating the identity on first credential insert - so existing code (including every earlier test fixture) needed no changes. registry.py: Licensor gains credential_label/rights; RIGHTS_TIERS + has_right() ordinal helper (enforcement is Control Plane's job, T03/ T04, not this task's); issue_sub_credential/revoke_sub_credential (revocation via a SECURITY DEFINER function, matching set_extension_status's existing pattern - trf_app has no UPDATE grant on licensors); authenticate() rejects revoked credentials identically to unrecognized ones. Attribution scoped honestly: ledger_entry.schema.json stays unmodified (frozen Stage 0 surface, additionalProperties:false) - per-entry human attribution is a hosting-layer-only column (ledger_entries.submitted_by_token, ledger.get_ledger_attribution()), recorded alongside but never inside the signed entry payload. Narrower than "the signature names the human," but exactly the "(or an accompanying attributable field)" alternative this task's own description anticipated. All four Docker-gated test files that append Ledger entries needed migration 0005 added (append_entry's INSERT now references the new column). New tests/test_licensor_credentials.py (8 tests): multi- credential resolution, duplicate-label rejection, revocation and its idempotence, invalid-rights rejection, the has_right helper, per-entry attribution recorded and not leaking into exported ledger JSON, and DB-level UPDATE rejection. Full suite: 84 offline, 41 with Docker (up from 30); no stray containers left running.
524 lines
22 KiB
Python
524 lines
22 KiB
Python
"""Integration tests for WP-0006-T04 (hosted Target Ledger append API).
|
|
|
|
Same ephemeral, disposable Postgres-via-Docker pattern as
|
|
test_registry_hosting.py (never the shared state-hub instance). The single
|
|
property this file tests most aggressively, per the task's own framing:
|
|
appending entries through the hosted API and then folding the exported
|
|
result offline (`fold.fold_outstanding_target`) must produce exactly the
|
|
same Outstanding Target a hosted metrics/attestation component would later
|
|
report — a hosted service that silently diverges from the offline fold is
|
|
the highest-risk defect this component could introduce.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import shutil
|
|
import subprocess
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
psycopg = pytest.importorskip("psycopg")
|
|
pytest.importorskip("fastapi")
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
|
|
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",
|
|
]
|
|
|
|
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-ledger-{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 conn:
|
|
for migration in MIGRATIONS:
|
|
conn.execute(migration.read_text(encoding="utf-8"))
|
|
conn.commit()
|
|
token = "test-token-acme"
|
|
conn.execute(
|
|
"INSERT INTO licensors (token, licensor_id) VALUES (%s, %s)",
|
|
(token, "acme-corp"),
|
|
)
|
|
other_token = "test-token-other"
|
|
conn.execute(
|
|
"INSERT INTO licensors (token, licensor_id) VALUES (%s, %s)",
|
|
(other_token, "other-corp"),
|
|
)
|
|
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,
|
|
"token": token,
|
|
"other_token": other_token,
|
|
}
|
|
finally:
|
|
subprocess.run(["docker", "stop", name], capture_output=True)
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(pg_container, monkeypatch):
|
|
monkeypatch.setenv("TRF_DATABASE_URL", pg_container["app_dsn"])
|
|
monkeypatch.setenv("TRF_SIGNING_KEY_HEX", "11" * 32)
|
|
from target_revenue.service import app as app_module
|
|
|
|
if hasattr(app_module.app.state, "pool"):
|
|
app_module.app.state.pool.close()
|
|
del app_module.app.state.pool
|
|
if hasattr(app_module.app.state, "signing_key"):
|
|
del app_module.app.state.signing_key
|
|
with TestClient(app_module.app) as c:
|
|
yield c
|
|
|
|
|
|
def auth_headers(token: str):
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@pytest.fixture()
|
|
def registered_phase(client, pg_container):
|
|
manifest = golden_manifest()
|
|
manifest["phase"]["id"] = manifest["phase"]["id"] + "-ledger-test-" + uuid.uuid4().hex[:6]
|
|
resp = client.post("/phases", json=manifest, headers=auth_headers(pg_container["token"]))
|
|
assert resp.status_code == 201, resp.text
|
|
return manifest
|
|
|
|
|
|
def _entry(phase_id: str, entry_id: str, entry_type: str, amount: float, recognized_at: str):
|
|
return {
|
|
"id": entry_id,
|
|
"phase": phase_id,
|
|
"type": entry_type,
|
|
"amount": amount,
|
|
"currency": "USD",
|
|
"recognized_at": recognized_at,
|
|
"evidence_reference": f"confidential:evidence:{entry_id}",
|
|
"extension": {"id": "trsl:extension:development-license", "version": "1.0"},
|
|
}
|
|
|
|
|
|
def test_append_and_hash_chain_links(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e1 = _entry(phase_id, "trsl:entry:ledgertest0001", "development-credit", 1000, "2026-08-01T00:00:00Z")
|
|
r1 = client.post(f"/phases/{phase_id}/ledger", json=e1, headers=auth_headers(pg_container["token"]))
|
|
assert r1.status_code == 201, r1.text
|
|
stored1 = r1.json()
|
|
assert stored1["previous_entry_hash"] == "GENESIS"
|
|
assert "signature" in stored1
|
|
|
|
e2 = _entry(phase_id, "trsl:entry:ledgertest0002", "development-credit", 500, "2026-08-02T00:00:00Z")
|
|
r2 = client.post(f"/phases/{phase_id}/ledger", json=e2, headers=auth_headers(pg_container["token"]))
|
|
assert r2.status_code == 201, r2.text
|
|
stored2 = r2.json()
|
|
|
|
from target_revenue import hashing
|
|
|
|
assert stored2["previous_entry_hash"] == hashing.entry_hash(stored1)
|
|
# Full chain, as exported, must independently verify.
|
|
hashing.verify_chain([stored1, stored2])
|
|
|
|
|
|
def test_client_cannot_supply_previous_entry_hash_or_signature(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
forged = _entry(phase_id, "trsl:entry:ledgertest0003", "development-credit", 1, "2026-08-01T00:00:00Z")
|
|
forged["previous_entry_hash"] = "GENESIS"
|
|
resp = client.post(f"/phases/{phase_id}/ledger", json=forged, headers=auth_headers(pg_container["token"]))
|
|
assert resp.status_code == 422
|
|
assert "must not be supplied" in resp.json()["detail"]
|
|
|
|
|
|
def test_other_licensor_cannot_append_to_this_phase(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e = _entry(phase_id, "trsl:entry:ledgertest0004", "development-credit", 1, "2026-08-01T00:00:00Z")
|
|
resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["other_token"]))
|
|
assert resp.status_code == 422
|
|
assert "not authorized" in resp.json()["detail"]
|
|
|
|
|
|
def test_currency_mismatch_rejected(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e = _entry(phase_id, "trsl:entry:ledgertest0005", "development-credit", 1, "2026-08-01T00:00:00Z")
|
|
e["currency"] = "EUR"
|
|
resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
|
|
assert resp.status_code == 422
|
|
assert "currency" in resp.json()["detail"]
|
|
|
|
|
|
def test_hosted_ledger_export_folds_identically_to_offline_computation(
|
|
client, pg_container, registered_phase
|
|
):
|
|
"""The property T04's task description calls out as highest-risk:
|
|
hosted append + export, folded offline, must match what was appended."""
|
|
from target_revenue import fold, hashing
|
|
|
|
phase_id = registered_phase["phase"]["id"]
|
|
plan = [
|
|
("development-credit", 25000),
|
|
("development-credit", 10000),
|
|
("remission-credit", 5000),
|
|
("credit-reversal", 2000),
|
|
]
|
|
for i, (entry_type, amount) in enumerate(plan, start=1):
|
|
e = _entry(
|
|
phase_id,
|
|
f"trsl:entry:ledgertestfold{i:04d}",
|
|
entry_type,
|
|
amount,
|
|
f"2026-08-0{i}T00:00:00Z",
|
|
)
|
|
if entry_type == "credit-reversal":
|
|
e["reverses"] = "trsl:entry:ledgertestfold0001"
|
|
resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
|
|
assert resp.status_code == 201, resp.text
|
|
|
|
export = client.get(f"/phases/{phase_id}/ledger")
|
|
assert export.status_code == 200
|
|
entries = export.json()
|
|
assert len(entries) == len(plan)
|
|
|
|
# Chain integrity survives the round trip through Postgres/JSONB/API.
|
|
hashing.verify_chain(entries)
|
|
|
|
result = fold.fold_outstanding_target(
|
|
registered_phase["phase"]["initial_target"]["amount"], entries
|
|
)
|
|
expected_development_credit = 25000 + 10000 - 2000
|
|
expected_remission_credit = 5000
|
|
assert result.development_credit == expected_development_credit
|
|
assert result.remission_credit == expected_remission_credit
|
|
assert result.outstanding_target == max(
|
|
0.0,
|
|
registered_phase["phase"]["initial_target"]["amount"]
|
|
- expected_development_credit
|
|
- expected_remission_credit,
|
|
)
|
|
|
|
|
|
def test_public_key_endpoint_unauthenticated_and_verifies_signature(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e = _entry(phase_id, "trsl:entry:ledgertestpk0001", "development-credit", 1, "2026-08-01T00:00:00Z")
|
|
stored = client.post(
|
|
f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"])
|
|
).json()
|
|
|
|
pk_resp = client.get("/public-key")
|
|
assert pk_resp.status_code == 200
|
|
assert pk_resp.json()["algorithm"] == "ed25519"
|
|
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
|
|
|
|
from target_revenue import hashing
|
|
|
|
public_key = Ed25519PublicKey.from_public_bytes(
|
|
bytes.fromhex(pk_resp.json()["public_key_hex"])
|
|
)
|
|
assert hashing.verify_record_signature(stored, stored["signature"], public_key)
|
|
|
|
|
|
def test_application_role_cannot_update_or_delete_ledger_entries(pg_container):
|
|
with psycopg.connect(pg_container["app_dsn"]) as conn:
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
conn.execute("UPDATE ledger_entries SET signature = 'x' WHERE entry_id = 'nonexistent'")
|
|
conn.rollback()
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
conn.execute("DELETE FROM ledger_entries WHERE entry_id = 'nonexistent'")
|
|
conn.rollback()
|
|
|
|
|
|
def test_duplicate_entry_id_rejected(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e = _entry(phase_id, "trsl:entry:ledgertestdup0001", "development-credit", 1, "2026-08-01T00:00:00Z")
|
|
first = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
|
|
assert first.status_code == 201
|
|
second = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
|
|
assert second.status_code == 422
|
|
assert "already exists" in second.json()["detail"]
|
|
|
|
|
|
def test_metrics_endpoint_unauthenticated_and_matches_offline_computation(
|
|
client, pg_container, registered_phase
|
|
):
|
|
"""WP-0006-T05: the hosted /metrics response must match what
|
|
target_revenue.metrics.compute_metrics computes offline from the same
|
|
exported Manifest + Ledger — the metrics-layer analogue of T04's
|
|
hosted/offline fold-agreement test."""
|
|
from target_revenue import metrics as metrics_module
|
|
|
|
phase_id = registered_phase["phase"]["id"]
|
|
e = _entry(phase_id, "trsl:entry:ledgertestmetrics0001", "development-credit", 4000, "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, resp.text
|
|
|
|
# No Authorization header at all: metrics are public per FR-9/FR-10.
|
|
metrics_resp = client.get(f"/phases/{phase_id}/metrics")
|
|
assert metrics_resp.status_code == 200
|
|
hosted = metrics_resp.json()
|
|
|
|
entries = client.get(f"/phases/{phase_id}/ledger").json()
|
|
as_of = metrics_module.datetime.fromisoformat(hosted["as_of"])
|
|
offline = metrics_module.compute_metrics(registered_phase, entries, as_of)
|
|
|
|
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()
|
|
|
|
|
|
def _breach_event(event_id: str, case_id: str, event_type: str, **overrides) -> dict:
|
|
event = {
|
|
"id": event_id,
|
|
"case_id": case_id,
|
|
"event_type": event_type,
|
|
"category": "unauthorized-commercial-use",
|
|
"event_at": "2026-08-01T00:00:00Z",
|
|
}
|
|
event.update(overrides)
|
|
return event
|
|
|
|
|
|
def test_breach_record_defaults_anonymized_and_lists_lifecycle_events(
|
|
client, pg_container, registered_phase
|
|
):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
case_id = "trsl:case:conformcase0001"
|
|
|
|
alleged = client.post(
|
|
f"/phases/{phase_id}/breach-records",
|
|
json=_breach_event("trsl:breach:conformcase00010001", case_id, "alleged"),
|
|
headers=auth_headers(pg_container["token"]),
|
|
)
|
|
assert alleged.status_code == 201, alleged.text
|
|
assert alleged.json()["anonymized"] is True
|
|
assert alleged.json()["named_entitlement_holder"] is None
|
|
assert "signature" in alleged.json()
|
|
|
|
determined = client.post(
|
|
f"/phases/{phase_id}/breach-records",
|
|
json=_breach_event(
|
|
"trsl:breach:conformcase00010002", case_id, "determined",
|
|
event_at="2026-08-15T00:00:00Z",
|
|
),
|
|
headers=auth_headers(pg_container["token"]),
|
|
)
|
|
assert determined.status_code == 201, determined.text
|
|
|
|
records = client.get(f"/phases/{phase_id}/breach-records")
|
|
assert records.status_code == 200
|
|
events = records.json()
|
|
assert [e["event_type"] for e in events] == ["alleged", "determined"]
|
|
# The "alleged" event must never have been edited by the "determined"
|
|
# publication — both remain, in order, as separate records.
|
|
assert events[0]["id"] == "trsl:breach:conformcase00010001"
|
|
assert events[1]["id"] == "trsl:breach:conformcase00010002"
|
|
|
|
|
|
def test_breach_record_named_disclosure_requires_explicit_cua_authorization(
|
|
client, pg_container, registered_phase
|
|
):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
unauthorized = client.post(
|
|
f"/phases/{phase_id}/breach-records",
|
|
json=_breach_event(
|
|
"trsl:breach:conformnamed0001", "trsl:case:conformnamed", "alleged",
|
|
anonymized=False, named_entitlement_holder="Acme Corp",
|
|
),
|
|
headers=auth_headers(pg_container["token"]),
|
|
)
|
|
assert unauthorized.status_code == 422
|
|
assert "named_disclosure_authorized_under_cua" in unauthorized.json()["detail"]
|
|
|
|
authorized = client.post(
|
|
f"/phases/{phase_id}/breach-records",
|
|
json=_breach_event(
|
|
"trsl:breach:conformnamed0002", "trsl:case:conformnamed", "alleged",
|
|
anonymized=False,
|
|
named_entitlement_holder="Acme Corp",
|
|
named_disclosure_authorized_under_cua=True,
|
|
),
|
|
headers=auth_headers(pg_container["token"]),
|
|
)
|
|
assert authorized.status_code == 201, authorized.text
|
|
assert authorized.json()["named_entitlement_holder"] == "Acme Corp"
|
|
|
|
|
|
def test_other_licensor_cannot_publish_breach_record_for_this_phase(
|
|
client, pg_container, registered_phase
|
|
):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
resp = client.post(
|
|
f"/phases/{phase_id}/breach-records",
|
|
json=_breach_event("trsl:breach:conformcross0001", "trsl:case:conformcross", "alleged"),
|
|
headers=auth_headers(pg_container["other_token"]),
|
|
)
|
|
assert resp.status_code == 422
|
|
assert "not authorized" in resp.json()["detail"]
|
|
|
|
|
|
def test_breach_record_signature_verifiable_with_public_key(client, pg_container, registered_phase):
|
|
phase_id = registered_phase["phase"]["id"]
|
|
published = client.post(
|
|
f"/phases/{phase_id}/breach-records",
|
|
json=_breach_event("trsl:breach:conformverify0001", "trsl:case:conformverify", "alleged"),
|
|
headers=auth_headers(pg_container["token"]),
|
|
).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_breach_records(pg_container):
|
|
with psycopg.connect(pg_container["app_dsn"]) as conn:
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
conn.execute("UPDATE breach_records SET category = 'x' WHERE record_id = 'nonexistent'")
|
|
conn.rollback()
|
|
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
|
conn.execute("DELETE FROM breach_records WHERE record_id = 'nonexistent'")
|
|
conn.rollback()
|