diff --git a/CHANGELOG.md b/CHANGELOG.md index d50216b..3404a7b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,32 @@ evidence for each increment is retained by the maintainers and summarised here. ## 2026-09-26 +### `slice/026`: Trinity keeps working when the link goes down + +A site that loses its connection keeps operating, and the record of what it did survives the gap intact. + +Every receipt now carries a clock that cannot run backwards, even if the machine's own clock does, so two +machines that worked separately can be lined up afterwards. The clock is part of what gets signed, which matters: +comparing two records by something anyone could edit afterwards would prove nothing. What the clock is worth is +written down plainly alongside it. It is the machine's own reading of the time and it is not trustworthy on its +own; a deployment that needs the time itself to be trustworthy supplies a trusted source, and the standards +register records that as something no version of this software can supply for you. + +Receipts that cannot be sent wait in a durable queue and are delivered in order when the link returns, exactly as +they were signed, never rebuilt. The queue has a limit, and reaching it stops Trinity accepting new work rather +than letting it act with no way to confirm anything. That refusal is itself recorded. + +When two machines reconnect, their records are compared by a single value rather than message by message. Where +they disagree, each disagreement is recorded and signed. **Neither record is ever rewritten.** Two machines that +both acted while unable to see each other produced two true accounts of what each did, and nothing available +afterwards turns those into one account of what happened; claiming otherwise would mean inventing the part that +cannot be known. + +Records written before this increment remain valid and verifiable, and a test confirms it by verifying one built +under the previous format. + +Verified: gate green at exit 0; 904 tests passing; coverage 80.31%, level with the previous increment. + ### `slice/125`: the credential filter cannot be switched off by a log line Trinity masks anything credential-shaped on its way into a log. That filter could be killed by diff --git a/bin/verify_receipt.exs b/bin/verify_receipt.exs index 8e272df..c935159 100755 --- a/bin/verify_receipt.exs +++ b/bin/verify_receipt.exs @@ -6,7 +6,7 @@ # (slice 024, AC7). Run it from any directory against a file `mix trinity.receipts.export` # wrote; the registry travels inside the file. # -# elixir verify_receipt.exs receipts.json [--schemes receipt_v2_ed25519,receipt_v2_p384] +# elixir verify_receipt.exs receipts.json [--schemes receipt_v3_ed25519,receipt_v2_ed25519] # # Exit codes, the vocabulary an operator learns once: 0 verified, 1 invalid, 2 usage, # 5 trust not established (a key id the registry does not know), 6 compromised key. @@ -22,10 +22,17 @@ # file and the in-app verifier agree on the same inputs. defmodule VerifyReceipt do + # Every scheme version this script can read. `v2` is slice 024's; `v3` is slice 026's and carries + # a hybrid logical clock in the signed payload. Both are listed because a chain that spans the + # bump has rows of each, and dropping the older one would mean a verifier that refuses bytes it + # signed itself last week. @families %{ "receipt_v2_ed25519" => {"ed25519", :eddsa, :none, :ed25519}, "receipt_v2_p384" => {"p384", :ecdsa, :sha384, :secp384r1}, - "receipt_v2_mldsa87" => {"mldsa87", :mldsa87, :none, nil} + "receipt_v2_mldsa87" => {"mldsa87", :mldsa87, :none, nil}, + "receipt_v3_ed25519" => {"ed25519", :eddsa, :none, :ed25519}, + "receipt_v3_p384" => {"p384", :ecdsa, :sha384, :secp384r1}, + "receipt_v3_mldsa87" => {"mldsa87", :mldsa87, :none, nil} } @signed_kinds ~w(decision effect boot cap) diff --git a/config/test.exs b/config/test.exs index 8c03d21..2ee2e1b 100644 --- a/config/test.exs +++ b/config/test.exs @@ -240,7 +240,12 @@ config :phoenix, # Slice 024: the receipt signing key and registry for the suite live under the project's # ignored tmp/, never in the data directory of the machine running the tests. -config :trinity, :receipts, keys_dir: Path.expand("../tmp/test_keys", __DIR__) +# Slice 026: the forwarder's timer is off in the suite. A background drain racing an assertion +# about queue depth is a flaky test about nothing; `Trinity.Receipts.Forwarder.drain/0` is called +# explicitly where a test wants a pass to happen. +config :trinity, :receipts, + keys_dir: Path.expand("../tmp/test_keys", __DIR__), + forward_interval_ms: 0 # Slice 032: the suite embeds with the deterministic fake; nothing leaves the machine and no # model is needed. diff --git a/docs/07-security-model.md b/docs/07-security-model.md index 2fb1519..678ba2c 100644 --- a/docs/07-security-model.md +++ b/docs/07-security-model.md @@ -442,6 +442,63 @@ The receipt names the fields that changed and never their values, and the `[:trinity, :tool, :surface_drift]` event carries the same. A description is content, and the point of holding the tool is that its new content has not been read by anyone entitled to approve it. +## Disconnected operation: the clock, the queue and the merge (Slice 026, as built) + +A field or clinical site loses its link and keeps working. Three mechanisms make that safe to do, and each has a +limit worth stating before its benefit. + +### The clock, and what it is not worth + +Every receipt written under `receipt_v3_*` carries a hybrid logical clock (Kulkarni et al., 2014): +`{wall, counter, node}`, in the **signed** payload. It is there because a hash chain orders its own rows +perfectly and says nothing about a chain that ran elsewhere at the same time, and a merge needs an order that +survives a partition. + +**A signed receipt with an untrusted clock is weaker evidence than a signed receipt with a trusted one, and this +clock is untrusted.** It is the host's own reading. A host that lies about the time produces receipts that are +internally consistent and wrong about when; the signature attests that *this host said so*, not that it was true. +A deployment that needs the *when* to be trustworthy supplies a trusted time source, and +`docs/09-standards-register.md` carries that as a real-world dependency rather than a tree property, because no +version of this tree can close it. + +What the clock does guarantee: + +- **It never goes backwards within a chain.** `Clock.next/2` is monotone even when the host's wall clock steps + back, and the writer refuses a row whose clock does not follow the tail. The refusal is itself receipted, with + both clocks named. +- **It is signed**, so the ordering claim cannot be edited by anything that cannot sign. This is why the clock is + in the signed payload rather than in `meta`, against R21's general rule that platform-set values go in `meta`: a + merge that ordered two chains by an unsigned field would order them by something any writer could rewrite, and + the merge would then be evidence of nothing. +- **It names the device**, so two chains are distinguishable and ties break deterministically. + +### The queue, and why it is bounded + +Receipts awaiting acknowledgement sit in a durable per-scope queue. The envelope is stored **byte for byte** and +never rebuilt at send time, so a retry hands the far side exactly the bytes that were signed; that is the +condition under which store-and-forward is invisible to a verifier checking a signature offline. + +Acknowledgement is in order, and a drain stops a scope at its first refusal rather than skipping past it. A far +side holding row 7 but not row 5 holds a gap it cannot see. + +**Past the bound, effects are denied rather than performed and left unacknowledged.** An unbounded queue turns a +long partition into an unbounded liability: the machine keeps acting, nothing can be confirmed, and the operator +finds out when the disk fills. The queue-full refusal is receipted and is deliberately not itself queued, because +a full queue that cannot record that it is full is a deadlock dressed as a safety property. + +### The merge, and what it refuses to do + +On reconnect, two chains are compared by Merkle tree head (RFC 6962 hashing: distinct leaf and node prefixes, +odd nodes promoted rather than duplicated, so two distinct chains cannot share a head). Equal heads mean every +row matches, at the cost of one comparison rather than one message per row. + +**A merge never rewrites either side.** It does not produce one reconciled chain. Both chains stay exactly as +they were and what the merge produces is a signed record of where they disagree, one conflict receipt per +differing position. This is the honest outcome rather than a limitation: two devices that both acted while unable +to see each other produced two true accounts of what each did, and nothing available afterwards turns those into +one account of what happened. A remote export that does not verify is refused before anything is compared, so +conflict receipts cannot be written into the local chain by handing the merge invented rows. + ## Data at rest - SQLite file under the OS data dir with 0600 perms. Optional at-rest encryption is a later slice (SQLCipher via exqlite build flag), noted rather than planned. diff --git a/docs/adr/0008-authority-is-an-adapter.md b/docs/adr/0008-authority-is-an-adapter.md index 9d1a8c3..24fb151 100644 --- a/docs/adr/0008-authority-is-an-adapter.md +++ b/docs/adr/0008-authority-is-an-adapter.md @@ -29,3 +29,38 @@ whether the thing you asked for is allowed to happen. - Slice 024 builds the behaviour, `Local`, the effect catalog and local receipts. No slice in this tree builds an adapter; that is downstream work in a downstream repository. - Slice 062 builds the MCP resource-server profile. It is scoped as identity and never as authority. + +## Amendment, 2026-09-26 (slice 026): a fifth callback, optional, for store-and-forward + +The Decision above names four callbacks. Store-and-forward adds a fifth, `forward_receipt/2`, and it is +**optional**, which is the part that matters for this ADR's posture: an adapter written against the four +published callbacks stays valid and unchanged, and an adapter that does not export the fifth is simply one this +machine never forwards to. + +The question this slice was blocked on was put to the external authority plane's maintainers on 2026-09-20 and +answered the same day, against their tree at `8bc693ee`. Their answer, and it shapes the contract: + +> store-and-forward does not change their side as long as the acknowledgement carries the envelope the plane +> signed, unmodified. Their verifiers check the plane's signature over the envelope bytes offline, so a queue can +> wrap, delay or re-deliver an envelope and cannot re-sign it or alter a leaf. + +So `forward_receipt/2` receives the exported receipt **byte for byte**. An implementation may not re-sign it, +rebuild it or alter a leaf. This tree holds up its end by storing the envelope at queue time rather than +reconstructing it at send time, and a test asserts the stored bytes equal the export; a queue that rebuilt the +envelope would break offline verification with nothing here failing. + +Two further consequences for the posture in point 5 above, that Trinity keeps no executor for effects an adapter +governs: + +- **The queue is bounded and the bound is a refusal.** Past it, effects are denied rather than performed and left + unacknowledged. A machine acting unboundedly while unable to confirm anything is exactly the state this ADR's + separation exists to avoid, and an unbounded queue would create it quietly. +- **A reconciliation row on the plane's side is wanted and was offered as small work on request.** It is not a + dependency: `AC1` to `AC4` run under `Local`, which acknowledges immediately, so the mode is proven in this + tree without any adapter. The request is made at this slice's G1, and its absence blocks nothing here. + +The merge this slice adds never rewrites either chain. It records where two chains disagree and signs that +record. That is deliberate and belongs in this ADR because it is a statement about what an authority boundary +can and cannot deliver: after a partition, two devices hold two true accounts of what each did, and nothing +available afterwards turns them into one account of what happened. A system that claimed otherwise would be +inventing the part it could not know. diff --git a/docs/receipt-scheme-mapping.md b/docs/receipt-scheme-mapping.md index d390e0d..7646d8c 100644 --- a/docs/receipt-scheme-mapping.md +++ b/docs/receipt-scheme-mapping.md @@ -24,6 +24,17 @@ emits is a SCITT Receipt, and this project makes no such claim. The name predate tree and is not being changed, because "receipt" is the right word for what it is to the person who owns the machine; but a reader who knows SCITT should not have to work this out for themselves. +## Two scheme versions, since slice 026 + +`receipt_v2_` is slice 024's and carries ten keys. `receipt_v3_` is slice 026's and +carries eleven: the ten plus `clock`, a hybrid logical clock for relating two chains that ran during +a partition. New rows are written under `v3`; `v2` rows are still verified under `v2`, row by row, +under each row's own scheme. + +That is a scheme bump rather than an edit: no byte signed under `v2` changed, and +`Trinity.Receipts.Signer.accepted_schemes/0` is what the verifier allows by default, so bumping the +version did not make previously signed rows unverifiable. The field list below is `v3`. + ## The envelope Trinity signs with today The signature is not over the canonical JSON directly. It is over DSSE's pre-authentication encoding @@ -33,11 +44,13 @@ a different one from SCITT's: DSSE rather than COSE. That matters for the last s ## The field mapping -Trinity's signed payload is RFC 8785 canonical JSON of exactly these ten keys +Trinity's signed payload is RFC 8785 canonical JSON of exactly these eleven keys (`Trinity.Receipts.ChainWriter`, the `body` map): `scheme`, `seq`, `chain_scope`, `prev_hash`, `kind`, `subject`, `decision`, `fingerprint`, `at`, -`key_id`. +`key_id`, `clock`. + +Under `receipt_v2_*` the same list without `clock`. `subject_ref` and `meta` are columns on the row but are **not** in the signed payload, so they are not covered by the signature and are not mapped below. @@ -51,6 +64,7 @@ not covered by the signature and are not mapped below. | `subject` | CWT `sub` (claim 2) | SCITT requires a single `tstr`; Trinity's signed `subject` is a map of references. Mapping it needs a canonical string form chosen, which is a decision, not a rename. | | *(absent)* | CWT `iss` (claim 1), **required** | Trinity has no issuer field. A single-operator machine has one issuer and it is implied by the key, which is why it was never needed and is exactly what a standards-track reader would look for first. | | `decision`, `fingerprint` | the statement payload | Domain content. SCITT does not constrain it. | +| `clock` | **no equivalent** | A hybrid logical clock (slice 026): `wall`, `counter`, `node`. SCITT has no notion of one because a transparency service supplies the ordering its receipts need. Trinity has no such service, so ordering across a partition has to travel in the statement, and it is signed for the reason the mapping document gives below: a merge that orders two chains by an unsigned field orders them by something any writer could rewrite. | | `seq`, `chain_scope`, `prev_hash` | **no equivalent** | These are the hash chain. In SCITT the equivalent guarantee comes from a Transparency Service's Verifiable Data Structure and is carried in a Receipt's proofs, not in the statement. Trinity's chain is the whole of its tamper-evidence, and it is local. | ## Why the field names are not being renamed @@ -76,7 +90,9 @@ do. - Trinity's receipts are **Signed Statements** in SCITT's sense, not Receipts. - Their integrity comes from a local hash chain with signed checkpoints, not from a transparency log. -- Every element of the signed payload maps to a published construct, and the three that do not map - (`seq`, `prev_hash`, `chain_scope`) are the ones SCITT delegates to a service Trinity does not run. +- Every element of the signed payload maps to a published construct except four: `seq`, `prev_hash` + and `chain_scope`, which are the ones SCITT delegates to a service Trinity does not run, and + `clock`, which exists for the same reason, since ordering across a partition is what a + transparency service would otherwise provide. - `subject_ref` and `meta` are unsigned row columns. Nothing should be inferred from them. - No conformance to RFC 9943 is claimed, and none should be inferred. diff --git a/lib/trinity/authority.ex b/lib/trinity/authority.ex index 3ee9d05..5665787 100644 --- a/lib/trinity/authority.ex +++ b/lib/trinity/authority.ex @@ -33,6 +33,23 @@ defmodule Trinity.Authority do @doc "Records a receipt of a kind with attributes; the adapter may forward it." @callback receipt(kind :: String.t(), attrs :: map()) :: {:ok, term()} | {:error, term()} + @doc """ + Hands a queued receipt envelope to the authority, byte for byte (slice 026). + + Optional. An implementation that does not export it is one this machine never forwards to, and + the queue for its scopes simply does not drain. + + The envelope is the exported receipt exactly as it was signed. An implementation **must not** + re-sign it, rebuild it or alter a leaf: the external authority plane's verifiers check the + signature over these bytes offline, which is the property that lets a queue wrap, delay and + re-deliver an envelope without the far side having to know a queue exists at all. + + `:ok` acknowledges. Any error leaves the entry pending, to be offered again. + """ + @callback forward_receipt(envelope :: map(), meta :: map()) :: :ok | {:error, term()} + + @optional_callbacks forward_receipt: 2 + @callbacks [stage: 2, decide: 3, execute: 3, receipt: 2] @doc "The callbacks every implementation must export, as `{name, arity}`." diff --git a/lib/trinity/authority/local.ex b/lib/trinity/authority/local.ex index a3854ea..99bf587 100644 --- a/lib/trinity/authority/local.ex +++ b/lib/trinity/authority/local.ex @@ -35,4 +35,15 @@ defmodule Trinity.Authority.Local do scope = Map.fetch!(attrs, :scope) Trinity.Receipts.append(scope, Map.put(attrs, :kind, kind)) end + + @doc """ + Acknowledges a queued envelope (slice 026). + + Standalone, there is no far side and nothing to send to, so this acknowledges immediately. It is + not a stub: it is what makes the local authority exercise the same queue-and-acknowledge path an + adapter does, so the mode is proven without one. A receipt still passes through the queue, is + still acknowledged in order, and the bound still applies. + """ + @impl true + def forward_receipt(envelope, _meta) when is_map(envelope), do: :ok end diff --git a/lib/trinity/receipts/chain_writer.ex b/lib/trinity/receipts/chain_writer.ex index 8ba364c..49db9ed 100644 --- a/lib/trinity/receipts/chain_writer.ex +++ b/lib/trinity/receipts/chain_writer.ex @@ -30,7 +30,17 @@ defmodule Trinity.Receipts.ChainWriter do import Ecto.Query - alias Trinity.Receipts.{Checkpoint, Envelope, KeyCustody, KeyRegistry, Receipt, Signer} + alias Trinity.Receipts.{ + Checkpoint, + Clock, + Envelope, + KeyCustody, + KeyRegistry, + Queue, + Receipt, + Signer + } + alias Trinity.Repo.Receipts, as: Repo require Logger @@ -41,6 +51,7 @@ defmodule Trinity.Receipts.ChainWriter do defstruct scope: nil, seq: 0, prev_hash: nil, + clock: nil, uncovered_first: nil, uncovered_count: 0, timer: nil @@ -121,8 +132,16 @@ defmodule Trinity.Receipts.ChainWriter do @impl true def handle_call({:append, attrs}, _from, state) do case do_append(attrs, state) do - {:ok, receipt, state} -> {:reply, {:ok, receipt}, maybe_checkpoint(receipt, state)} - {:error, reason} -> {:reply, {:error, reason}, state} + {:ok, receipt, state} -> + {:reply, {:ok, receipt}, maybe_checkpoint(receipt, state)} + + # A refusal that was itself receipted: the reply is the error, and the state advances, + # because the refusal row is in the chain and the tail moved. + {:error, reason, state} -> + {:reply, {:error, reason}, state} + + {:error, reason} -> + {:reply, {:error, reason}, state} end end @@ -166,11 +185,118 @@ defmodule Trinity.Receipts.ChainWriter do unless kind in Receipt.kinds(), do: raise(ArgumentError, "unknown receipt kind #{inspect(kind)}") - with {:ok, %{scheme: scheme, key_id: key_id}} <- selection(), - {row, at} = build_row(attrs, kind, scheme, key_id, state), + forward? = Map.get(attrs, :forward, true) + + with :ok <- check_queue_bound(forward?, state), + {:ok, %{scheme: scheme, key_id: key_id}} <- selection(), + {row, at, clock} = build_row(attrs, kind, scheme, key_id, state), + :ok <- check_clock(clock, state), bytes = Envelope.pae(Envelope.receipt_type(scheme), row.signed_payload), {:ok, signature} <- sign_if_needed(kind, bytes) do - insert_row(%{row | signature: signature, inserted_at: at}, state) + insert_row(%{row | signature: signature, inserted_at: at}, state, clock, forward?) + else + {:error, {:clock_regression, offered, held}} -> + refuse_clock_regression(offered, held, state) + + {:error, {:queue_full, depth, bound}} -> + refuse_queue_full(depth, bound, state) + + other -> + other + end + end + + # Slice 026, AC4. The bound is a refusal, not a buffer: past it the effect is denied rather than + # performed and left unacknowledged. An unbounded queue turns a long partition into an unbounded + # liability, where the machine keeps acting, nothing can be confirmed, and the operator finds out + # when the disk fills. + defp check_queue_bound(false, _state), do: :ok + + defp check_queue_bound(true, state) do + depth = Queue.depth(state.scope) + bound = Queue.bound() + if depth >= bound, do: {:error, {:queue_full, depth, bound}}, else: :ok + end + + defp refuse_queue_full(depth, bound, state) do + attrs = %{ + kind: "decision", + subject: %{ + "chain_scope" => state.scope, + "queue_depth" => depth, + "queue_bound" => bound + }, + decision: %{ + "outcome" => "refused", + "basis" => "queue", + "reason" => + "the outbound receipt queue is at its bound: refusing the effect rather than " <> + "performing it and leaving it unacknowledged" + }, + subject_ref: "queue_full:#{state.scope}:#{state.seq}", + # Not itself queued. If it were, a full queue could not record that it was full, which is a + # deadlock dressed as a safety property. + forward: false + } + + case do_append_unchecked(attrs, state) do + {:ok, _receipt, state} -> {:error, {:queue_full, depth, bound}, state} + {:error, why} -> {:error, {:queue_full_unrecorded, why}} + end + end + + # Slice 026, AC3. The writer's own clock cannot go backwards: `Clock.next/2` is monotone even + # when the host's wall clock is not. This guards the other way in, which is a clock supplied with + # the attributes, as the merge path and an adapter's replay both do. A row that claims to precede + # one already in the chain is refused, and the refusal is itself a receipt, because a refusal that + # leaves no record is indistinguishable from the call never having been made. + defp check_clock(nil, _state), do: :ok + defp check_clock(_clock, %{clock: nil}), do: :ok + + defp check_clock(clock, %{clock: held}) do + if Clock.compare(clock, held) == :gt, + do: :ok, + else: {:error, {:clock_regression, clock, held}} + end + + defp refuse_clock_regression(offered, held, state) do + attrs = %{ + kind: "decision", + subject: %{ + "chain_scope" => state.scope, + "offered_clock" => Clock.to_map(offered), + "held_clock" => Clock.to_map(held) + }, + decision: %{ + "outcome" => "refused", + "basis" => "clock", + "reason" => + "the offered clock does not follow the chain tail: a receipt may not claim to precede " <> + "a row already written" + }, + subject_ref: "clock_regression:#{state.scope}:#{state.seq}" + } + + # The refusal takes a fresh clock of its own, which is monotone by construction, so recording a + # refusal can never itself be refused. + case do_append_unchecked(attrs, state) do + {:ok, _receipt, state} -> {:error, {:clock_regression, offered, held}, state} + {:error, why} -> {:error, {:clock_regression_unrecorded, why}} + end + end + + defp do_append_unchecked(attrs, state) do + with {:ok, %{scheme: scheme, key_id: key_id}} <- selection(), + {row, at, clock} = + build_row(Map.delete(attrs, :clock), "decision", scheme, key_id, state), + bytes = Envelope.pae(Envelope.receipt_type(scheme), row.signed_payload), + {:ok, signature} <- sign_if_needed("decision", bytes) do + insert_row( + %{row | signature: signature, inserted_at: at}, + state, + clock, + Map.get(attrs, :forward, true) + ) end end @@ -178,18 +304,27 @@ defmodule Trinity.Receipts.ChainWriter do seq = state.seq + 1 at = DateTime.utc_now() - body = %{ - "scheme" => scheme, - "seq" => seq, - "chain_scope" => state.scope, - "prev_hash" => state.prev_hash, - "kind" => kind, - "subject" => Map.get(attrs, :subject, %{}), - "decision" => Map.get(attrs, :decision), - "fingerprint" => Map.get(attrs, :fingerprint), - "at" => DateTime.to_iso8601(at), - "key_id" => key_id - } + clock = + cond do + not Signer.clocked?(scheme) -> nil + offered = Map.get(attrs, :clock) -> offered + true -> Clock.next(state.clock) + end + + body = + %{ + "scheme" => scheme, + "seq" => seq, + "chain_scope" => state.scope, + "prev_hash" => state.prev_hash, + "kind" => kind, + "subject" => Map.get(attrs, :subject, %{}), + "decision" => Map.get(attrs, :decision), + "fingerprint" => Map.get(attrs, :fingerprint), + "at" => DateTime.to_iso8601(at), + "key_id" => key_id + } + |> then(fn b -> if clock, do: Map.put(b, "clock", Clock.to_map(clock)), else: b end) payload = Envelope.canonical(body) hash = Envelope.hash(Envelope.pae(Envelope.receipt_type(scheme), payload)) @@ -208,17 +343,36 @@ defmodule Trinity.Receipts.ChainWriter do meta: Map.get(attrs, :meta, %{}) } - {row, at} + {row, at, clock} end - defp insert_row(%Receipt{} = row, state) do - case Repo.insert(row) do + # The receipt and its queue entry land together or not at all. A receipt in the chain with no + # queue entry is one the far side will never see and nothing will ever notice; a queue entry with + # no receipt is a promise about a row that does not exist. + defp insert_row(%Receipt{} = row, state, clock, forward?) do + case Repo.transaction(fn -> insert_and_queue(row, forward?) end) do {:ok, receipt} -> - state = %{state | seq: receipt.seq, prev_hash: receipt.receipt_hash} + state = %{state | seq: receipt.seq, prev_hash: receipt.receipt_hash, clock: clock} {:ok, receipt, track_uncovered(receipt, state)} - {:error, changeset} -> - {:error, {:insert, changeset.errors}} + {:error, reason} -> + {:error, reason} + end + end + + defp insert_and_queue(row, forward?) do + case Repo.insert(row) do + {:ok, receipt} -> queue_entry(receipt, forward?) + {:error, changeset} -> Repo.rollback({:insert, changeset.errors}) + end + end + + defp queue_entry(receipt, false), do: receipt + + defp queue_entry(receipt, true) do + case Repo.insert(Queue.entry_changeset(receipt)) do + {:ok, _entry} -> receipt + {:error, changeset} -> Repo.rollback({:queue_insert, changeset.errors}) end end @@ -347,6 +501,19 @@ defmodule Trinity.Receipts.ChainWriter do ## Rehydrate + # The tail's clock is read back so the chain stays monotone across a restart. A `v2` tail has + # none, and a chain that crosses the bump simply starts its clock at the first `v3` row. + defp tail_clock(nil), do: nil + + defp tail_clock(%Receipt{signed_payload: payload}) do + with {:ok, body} <- JSON.decode(payload), + {:ok, clock} <- Clock.from_map(body["clock"]) do + clock + else + _ -> nil + end + end + defp rehydrate(scope) do tail = Repo.one( @@ -369,6 +536,7 @@ defmodule Trinity.Receipts.ChainWriter do scope: scope, seq: (tail && tail.seq) || 0, prev_hash: tail && tail.receipt_hash, + clock: tail_clock(tail), uncovered_first: first, uncovered_count: count }} diff --git a/lib/trinity/receipts/clock.ex b/lib/trinity/receipts/clock.ex new file mode 100644 index 0000000..f1ecb8e --- /dev/null +++ b/lib/trinity/receipts/clock.ex @@ -0,0 +1,172 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Receipts.Clock do + @moduledoc """ + A hybrid logical clock stamped on every receipt from slice 026. + + ## Why a chain needs one at all + + Within one chain, `seq` and `prev_hash` already order every row, and they order it better than + any clock can: the order is cryptographic rather than asserted. The clock is not for that. It is + for the one thing a hash chain cannot do, which is relate **two** chains that ran at the same time + on different machines without either being able to see the other. A field site that loses its link + keeps working; when it reconnects, two chains have to be compared, and comparison needs an order + that survives a partition. + + A wall clock alone cannot do it: two machines disagree, and a machine whose clock steps backwards + produces receipts that appear to precede ones it already wrote. A Lamport counter alone cannot do + it either: it orders events but says nothing a human or an auditor can place in a day. A hybrid + logical clock (Kulkarni et al., 2014) carries both, taking the wall reading when it is ahead and + incrementing a counter when it is not, so it never goes backwards and stays close to real time. + + ## What it is worth as evidence, stated plainly + + **A signed receipt with an untrusted clock is weaker evidence than a signed receipt with a trusted + one, and this clock is untrusted.** It is the host's own reading. A host that lies about the time + produces receipts that are internally consistent and wrong about when, and no amount of signing + fixes that, because the signature attests that *this host said so*, not that it was true. + + What the clock does guarantee, and it is worth having: + + - **It never goes backwards within a chain.** `next/2` is monotone by construction, and the writer + refuses a row whose clock is not greater than its predecessor's, so a host that steps its wall + clock back cannot produce a receipt that appears to precede one it already wrote. The refusal is + itself receipted. + - **It is signed**, so the ordering claim cannot be edited afterwards by anything that cannot sign. + That is why the clock is in the signed payload and not in `meta`: a merge that orders two chains + by an unsigned field orders them by something any writer could rewrite, which would make the + merge evidence of nothing. + - **It names the device**, so two chains that ran concurrently are distinguishable and a tie is + broken deterministically rather than arbitrarily. + + A deployment that needs the *when* to be trustworthy supplies a trusted time source. That is a + deployment matter and `docs/09-standards-register.md` carries it as a real-world dependency rather + than a tree property, because no version of this tree can close it. + """ + + # Sobelow reads `@sobelow_skip` out of the source AST; the compiler never sees it used, so it + # has to be registered or every annotated module warns. Same as `Trinity.Paths`. + Module.register_attribute(__MODULE__, :sobelow_skip, persist: true) + + @type t :: %__MODULE__{wall: non_neg_integer(), counter: non_neg_integer(), node: String.t()} + + @enforce_keys [:wall, :counter, :node] + defstruct [:wall, :counter, :node] + + @doc """ + The next clock after `prev`, for this device. + + The hybrid rule: take the wall reading if it is ahead of the last logical reading and reset the + counter; otherwise keep the logical reading and increment the counter. Either way the result is + strictly greater than `prev` for this node, so a chain's clocks never go backwards even when the + host's wall clock does. + + `prev` is `nil` for the first row of a chain. + """ + @spec next(t() | nil, non_neg_integer()) :: t() + def next(prev, wall_ms \\ System.system_time(:millisecond)) + + def next(nil, wall_ms), do: %__MODULE__{wall: wall_ms, counter: 0, node: node_id()} + + def next(%__MODULE__{} = prev, wall_ms) do + if wall_ms > prev.wall do + %__MODULE__{wall: wall_ms, counter: 0, node: node_id()} + else + %__MODULE__{wall: prev.wall, counter: prev.counter + 1, node: node_id()} + end + end + + @doc """ + Orders two clocks: `:lt`, `:eq` or `:gt`. + + Wall first, then counter, then the device id. The device id is a tiebreak and nothing more: two + receipts from different devices with the same wall and counter are genuinely concurrent, and this + gives them a stable order so that a merge is deterministic rather than dependent on which side + ran first. `concurrent?/2` is the honest question to ask about two clocks from different devices. + """ + @spec compare(t(), t()) :: :lt | :eq | :gt + def compare(%__MODULE__{} = a, %__MODULE__{} = b) do + cond do + a.wall < b.wall -> :lt + a.wall > b.wall -> :gt + a.counter < b.counter -> :lt + a.counter > b.counter -> :gt + a.node < b.node -> :lt + a.node > b.node -> :gt + true -> :eq + end + end + + @doc """ + True when two clocks are from different devices and neither strictly precedes the other in a way + this clock can witness. + + A hybrid logical clock carries no causal history, so the only concurrency it can report is + "different devices, same reading". Two receipts on different devices with different wall readings + are ordered by `compare/2` and that order is an assertion about two untrusted clocks, not a proof + of causality. The merge in slice 026 uses this to decide what to record as a conflict rather than + to decide what actually happened first, which it cannot know. + """ + @spec concurrent?(t(), t()) :: boolean() + def concurrent?(%__MODULE__{} = a, %__MODULE__{} = b), + do: a.node != b.node and a.wall == b.wall and a.counter == b.counter + + @doc "The clock as it appears in a signed payload: string keys, integers, no struct." + @spec to_map(t()) :: map() + def to_map(%__MODULE__{} = c), + do: %{"wall" => c.wall, "counter" => c.counter, "node" => c.node} + + @doc "Reads a clock back from a signed payload. `:error` when the shape is not one." + @spec from_map(term()) :: {:ok, t()} | :error + def from_map(%{"wall" => w, "counter" => c, "node" => n}) + when is_integer(w) and w >= 0 and is_integer(c) and c >= 0 and is_binary(n), + do: {:ok, %__MODULE__{wall: w, counter: c, node: n}} + + def from_map(_), do: :error + + @doc """ + This installation's device id, created once and then read. + + Stable across restarts because a merge has to be able to tell one machine's chain from another's + across a partition that may outlast several of them. It is a random identifier and names nothing + about the host: it exists to be different from other devices, not to describe this one. + """ + @spec node_id() :: String.t() + def node_id do + case :persistent_term.get({__MODULE__, :node_id}, nil) do + nil -> + id = read_or_create_node_id() + :persistent_term.put({__MODULE__, :node_id}, id) + id + + id -> + id + end + end + + @doc false + @spec forget_node_id() :: :ok + def forget_node_id, do: :persistent_term.erase({__MODULE__, :node_id}) && :ok + + # One function rather than two so the annotation covers the read and the write together, and + # public because the attribute only attaches to a public definition. + @doc false + # sobelow_skip reason: Traversal.FileModule, as `Trinity.Paths.keys_dir/0` itself: the path is + # the keys directory plus a constant, never input. + @sobelow_skip ["Traversal.FileModule"] + @spec read_or_create_node_id() :: String.t() + def read_or_create_node_id do + path = Path.join(Trinity.Paths.keys_dir(), "device_id") + + with {:ok, contents} <- File.read(path), + id when id != "" <- String.trim(contents) do + id + else + _ -> + id = 12 |> :crypto.strong_rand_bytes() |> Base.url_encode64(padding: false) + File.write!(path, id) + File.chmod!(path, 0o600) + id + end + end +end diff --git a/lib/trinity/receipts/forwarder.ex b/lib/trinity/receipts/forwarder.ex new file mode 100644 index 0000000..17379a9 --- /dev/null +++ b/lib/trinity/receipts/forwarder.ex @@ -0,0 +1,156 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Receipts.Forwarder do + @moduledoc """ + Drains the outbound receipt queue to the authority in force (slice 026). + + One pass, on a timer: for every scope with pending entries, offer them **oldest first** and stop + that scope at the first refusal. Stopping rather than skipping is the whole point. Receipts are a + chain, and a far side that has row 7 but not row 5 holds a gap it cannot see; offering 7 after 5 + failed would manufacture exactly that. So a scope drains in order or it waits. + + ## Being unreachable is not an error condition here + + A partition is the case this exists for, not an exception to it. An offer that fails leaves the + entry pending, increments its attempt count, records the reason, and the next pass tries again. + Nothing is dropped, nothing is retried in a tight loop, and no alarm is raised for a link being + down, because a field site losing its link is ordinary and the queue is the designed response to + it. What is not ordinary is the queue reaching its bound, and that refuses effects at the writer + rather than here. + + An authority that does not export `forward_receipt/2` is one this machine never forwards to; its + scopes simply do not drain, and that is a configuration fact rather than a failure. + """ + use GenServer + + alias Trinity.Receipts.Queue + + require Logger + + @default_interval_ms 5_000 + @batch 50 + + @doc false + def start_link(opts \\ []), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__) + + @doc """ + Runs one drain pass now and returns what it did. + + Exposed because a timer is not a thing a test should wait on, and because an operator wants a way + to ask "go now" after a link comes back rather than waiting out the interval. + """ + @spec drain_now() :: %{acked: non_neg_integer(), failed: non_neg_integer()} + def drain_now, do: GenServer.call(__MODULE__, :drain, 30_000) + + @doc """ + One drain pass over every scope with pending entries, without needing the process. + + Used by `drain_now/0` and by tests. + """ + @spec drain() :: %{acked: non_neg_integer(), failed: non_neg_integer()} + def drain do + Queue.scopes_with_pending() + |> Enum.reduce(%{acked: 0, failed: 0}, fn scope, acc -> + drain_scope(scope, acc) + end) + end + + @doc """ + One drain pass over a single scope. + + Separate from `drain/0` because "this link came back" is a fact about one scope, and draining + every other scope on the strength of it would offer receipts to adapters that are still down and + charge each of them an attempt. + """ + @spec drain(String.t()) :: %{acked: non_neg_integer(), failed: non_neg_integer()} + def drain(scope) when is_binary(scope), do: drain_scope(scope, %{acked: 0, failed: 0}) + + @impl true + def init(opts) do + interval = Keyword.get(opts, :interval_ms, interval_ms()) + if interval > 0, do: Process.send_after(self(), :tick, interval) + {:ok, %{interval: interval}} + end + + @impl true + def handle_call(:drain, _from, state), do: {:reply, drain(), state} + + @impl true + def handle_info(:tick, state) do + drain() + if state.interval > 0, do: Process.send_after(self(), :tick, state.interval) + {:noreply, state} + end + + def handle_info(_msg, state), do: {:noreply, state} + + defp drain_scope(scope, acc) do + scope + |> Queue.pending(@batch) + |> Enum.reduce_while(acc, fn entry, acc -> offer_one(scope, entry, acc) end) + end + + defp offer_one(scope, entry, acc) do + case offer(entry) do + :ok -> + record_ack(scope, entry, acc) + + {:error, reason} -> + Queue.fail(entry, reason) + {:halt, %{acc | failed: acc.failed + 1}} + end + end + + defp record_ack(scope, entry, acc) do + case Queue.ack(scope, entry.receipt_hash) do + :ok -> + {:cont, %{acc | acked: acc.acked + 1}} + + {:error, reason} -> + # The far side took it and this side could not record that. Stop: acknowledging anything + # later would close a gap that is now real. + Logger.warning("receipts: #{scope}: forwarded but not acked: #{inspect(reason)}") + {:halt, %{acc | failed: acc.failed + 1}} + end + end + + defp offer(entry) do + module = Trinity.Authority.impl() + + # `Code.ensure_loaded?/1` first: `function_exported?/3` answers false for a module that simply + # has not been loaded yet, which in a lazily-loading runtime is most of them. Without it, the + # very first drain after boot decides the authority cannot forward and never asks again. + if Code.ensure_loaded?(module) and function_exported?(module, :forward_receipt, 2) do + case JSON.decode(entry.envelope) do + {:ok, envelope} -> + safe_forward(module, envelope, %{ + "chain_scope" => entry.chain_scope, + "seq" => entry.seq, + "kind" => entry.kind, + "attempts" => entry.attempts + }) + + {:error, reason} -> + {:error, {:envelope_not_json, reason}} + end + else + {:error, {:no_forward_callback, module}} + end + end + + # An adapter is other people's code reached over a link that is by assumption unreliable. A raise + # from it is a failed delivery, not a reason to take this process down and stop draining every + # other scope. + defp safe_forward(module, envelope, meta) do + module.forward_receipt(envelope, meta) + rescue + e -> {:error, {:forward_raised, Exception.message(e)}} + catch + kind, reason -> {:error, {:forward_exited, kind, reason}} + end + + defp interval_ms do + Application.get_env(:trinity, :receipts, []) + |> Keyword.get(:forward_interval_ms, @default_interval_ms) + end +end diff --git a/lib/trinity/receipts/merge.ex b/lib/trinity/receipts/merge.ex new file mode 100644 index 0000000..f574689 --- /dev/null +++ b/lib/trinity/receipts/merge.ex @@ -0,0 +1,152 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Receipts.Merge do + @moduledoc """ + Reconciles this machine's chain with another device's after a partition (slice 026). + + ## What a merge is allowed to do, which is almost nothing + + **It never rewrites either side.** Not the local rows, not the remote export. A receipt is + evidence, and evidence that a later process can edit to make two accounts agree is not evidence. + So this does not merge in the sense of producing one reconciled chain: both chains stay exactly + as they were, and what the merge produces is a **record of where they disagree**. + + That is the honest outcome. Two devices that both acted while unable to see each other produced + two true accounts of what each did. Nothing available afterwards can turn those into one account + of what happened, and a system that claimed to would be inventing the part it could not know. + What an auditor needs is to be told precisely where the accounts part company, and for that + statement to be signed. + + ## What it records + + For each differing position, one conflict receipt naming the position, both hashes, and both + devices. One receipt per divergence rather than one for the whole run, because an auditor asking + what happened at row 9 deserves an answer about row 9. + + Identical chains produce nothing at all: a merge that writes a receipt saying "no conflict" every + time two devices sync turns the chain into a log of its own housekeeping. + + ## What it refuses + + A remote export that does not verify is refused before anything is compared. Recording a conflict + against an unverifiable chain would be recording that this machine disagrees with an assertion + nobody made, and it would let anyone who can reach the merge entry point write conflict receipts + into the local chain by handing it invented rows. + """ + + alias Trinity.Receipts + alias Trinity.Receipts.{Merkle, Verifier} + + @typedoc "What a merge did." + @type outcome :: + {:ok, :identical, map()} + | {:ok, :diverged, map()} + | {:error, term()} + + @doc """ + Compares the local chain for `scope` with a remote export and records the divergences. + + The remote export is the same shape `Trinity.Receipts.export/1` produces, from the other device. + It is verified first, and refused if it does not verify. + + Returns `{:ok, :identical, summary}` when the tree heads match, or `{:ok, :diverged, summary}` + with a conflict receipt written for each differing position. The summary carries both heads and + the positions, so a caller can report without re-reading the chain. + """ + @spec merge(String.t(), map()) :: outcome() + def merge(scope, remote_export) when is_binary(scope) and is_map(remote_export) do + with :ok <- verify_remote(remote_export) do + local_hashes = scope |> Receipts.list() |> Enum.map(& &1.receipt_hash) + remote_rows = remote_export["receipts"] |> Enum.sort_by(& &1["seq"]) + remote_hashes = Enum.map(remote_rows, & &1["receipt_hash"]) + + summary = %{ + local_head: Merkle.root(local_hashes), + remote_head: Merkle.root(remote_hashes), + local_rows: length(local_hashes), + remote_rows: length(remote_hashes) + } + + case Merkle.compare(local_hashes, remote_hashes) do + :identical -> + {:ok, :identical, Map.put(summary, :conflicts, [])} + + {:diverged, positions} -> + record_conflicts(scope, positions, local_hashes, remote_rows, summary) + end + end + end + + @doc """ + This machine's tree head for a scope: one value that answers "are we the same?". + + Sent to the other side instead of the chain. Equal heads mean every row matches, and that costs + one comparison rather than one message per row, which is the point of doing this with a tree on a + link that has just come back. + """ + @spec head(String.t()) :: String.t() + def head(scope), do: scope |> Receipts.list() |> Enum.map(& &1.receipt_hash) |> Merkle.root() + + defp verify_remote(export) do + case Verifier.verify(export) do + {:ok, _} -> :ok + {:error, code, reason} -> {:error, {:remote_does_not_verify, code, reason}} + end + end + + defp record_conflicts(scope, positions, local_hashes, remote_rows, summary) do + results = + Enum.map(positions, fn position -> + local = Enum.at(local_hashes, position - 1) + remote_row = Enum.at(remote_rows, position - 1) + + Receipts.append(scope, %{ + kind: "decision", + subject: %{ + "chain_scope" => scope, + "seq" => position, + "local_receipt_hash" => local, + "remote_receipt_hash" => remote_row && remote_row["receipt_hash"], + "local_head" => summary.local_head, + "remote_head" => summary.remote_head, + "remote_device" => device_of(remote_row) + }, + decision: %{ + "outcome" => "conflict", + "basis" => "merge", + "reason" => reason_for(local, remote_row) + }, + subject_ref: "merge_conflict:#{scope}:#{position}" + }) + end) + + case Enum.find(results, &match?({:error, _}, &1)) do + nil -> + {:ok, :diverged, Map.put(summary, :conflicts, positions)} + + {:error, why} -> + {:error, {:conflict_not_recorded, why}} + end + end + + defp reason_for(nil, _remote), + do: "the remote chain has a row at this position and this one does not" + + defp reason_for(_local, nil), + do: "this chain has a row at this position and the remote does not" + + defp reason_for(_local, _remote), + do: "both chains have a row at this position and they are not the same row" + + defp device_of(nil), do: nil + + defp device_of(row) do + with payload when is_binary(payload) <- row["signed_payload"], + {:ok, body} <- JSON.decode(payload), + %{"node" => node} <- body["clock"] do + node + else + _ -> nil + end + end +end diff --git a/lib/trinity/receipts/merkle.ex b/lib/trinity/receipts/merkle.ex new file mode 100644 index 0000000..08a8bdb --- /dev/null +++ b/lib/trinity/receipts/merkle.ex @@ -0,0 +1,96 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Receipts.Merkle do + @moduledoc """ + A Merkle tree over a chain's receipt hashes, for comparing two chains cheaply (slice 026). + + ## Why a tree when there is already a hash chain + + A hash chain already makes any alteration detectable, but only to someone holding the whole + chain. Comparing two chains by walking them costs a message per row, and the case this exists for + is a site that has just come back after a partition and has a link worth conserving. A tree head + is one value: equal heads mean two chains are the same for every row, and that answer costs one + comparison rather than thousands. + + ## The hashing is RFC 6962's, deliberately + + Certificate Transparency's rule: a leaf is `SHA-256(0x00 || data)` and an interior node is + `SHA-256(0x01 || left || right)`. The distinct prefixes are the whole point and are not + decoration. Without them a tree cannot tell a leaf from an interior node, and an attacker can + present an interior node's hash as though it were a leaf, which is the second-preimage attack + RFC 6962 section 2.1 exists to prevent. + + An odd node is **promoted**, not duplicated. Duplicating it is the Bitcoin construction and it + admits distinct leaf sets with the same root (CVE-2012-2459), which would mean two genuinely + different chains comparing as identical. That is the one thing this module must never do. + + An empty chain hashes to `SHA-256("")`, RFC 6962's rule for an empty tree, so "no rows" is a + value that can be compared rather than a special case every caller has to handle. + """ + + @doc """ + The tree head over an ordered list of receipt hashes. + + Order matters: these are the rows of a chain in `seq` order, and two chains holding the same rows + in a different order are not the same chain. + """ + @spec root([String.t()]) :: String.t() + def root([]), do: :crypto.hash(:sha256, "") |> Base.encode16(case: :lower) + + def root(hashes) when is_list(hashes) do + hashes + |> Enum.map(&leaf/1) + |> build() + |> Base.encode16(case: :lower) + end + + @doc """ + Compares two chains by their rows. + + `:identical` when every row matches. `{:diverged, seqs}` otherwise, where `seqs` are the + one-based positions that differ, including positions present on one side and not the other. + + The positions are returned rather than only the first, because a merge has to record every + divergence as its own conflict: "these chains differ from row 4 onwards" is one fact where there + may be thirty, and an auditor asking what happened at row 9 deserves a receipt about row 9. + """ + @spec compare([String.t()], [String.t()]) :: :identical | {:diverged, [pos_integer()]} + def compare(local, remote) when is_list(local) and is_list(remote) do + if root(local) == root(remote) do + :identical + else + {:diverged, differing_positions(local, remote)} + end + end + + @doc "The leaf hash of one value, RFC 6962's `SHA-256(0x00 || data)`." + @spec leaf(String.t()) :: binary() + def leaf(data) when is_binary(data), do: :crypto.hash(:sha256, <<0>> <> data) + + @doc "The interior node hash of two children, RFC 6962's `SHA-256(0x01 || left || right)`." + @spec node_hash(binary(), binary()) :: binary() + def node_hash(left, right) when is_binary(left) and is_binary(right), + do: :crypto.hash(:sha256, <<1>> <> left <> right) + + defp build([single]), do: single + + defp build(nodes) do + nodes + |> Enum.chunk_every(2) + |> Enum.map(fn + [left, right] -> node_hash(left, right) + # Promoted, never duplicated: duplicating an odd node lets two distinct leaf sets share a + # root (CVE-2012-2459), which here would mean two different chains comparing as identical. + [only] -> only + end) + |> build() + end + + defp differing_positions(local, remote) do + len = max(length(local), length(remote)) + + for i <- 0..(len - 1)//1, + Enum.at(local, i) != Enum.at(remote, i), + do: i + 1 + end +end diff --git a/lib/trinity/receipts/queue.ex b/lib/trinity/receipts/queue.ex new file mode 100644 index 0000000..b4f9619 --- /dev/null +++ b/lib/trinity/receipts/queue.ex @@ -0,0 +1,169 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Receipts.Queue do + @moduledoc """ + The durable outbound queue: receipts awaiting acknowledgement by the authority adapter (slice 026). + + ## What it does and does not do + + It **carries envelopes and never re-signs them.** The stored `envelope` is the exported receipt + byte for byte, so a delay, a retry or a re-delivery hands the adapter exactly the bytes that were + signed. That is not an implementation preference: the external authority plane answered the + question that blocked this slice by saying store-and-forward is invisible to them precisely + because their verifiers check a signature over those bytes offline, and a queue that rebuilt an + envelope at send time would break that without anything failing loudly. + + It **acknowledges in order.** A gap in acknowledgements is a gap in what the far side has seen, + and an out-of-order ack would silently close one. `ack/2` refuses a receipt whose predecessors in + the same scope are still pending. + + It is **bounded**, and the bound is a refusal rather than a buffer. Past the bound the writer + refuses new forwardable receipts, so effects are denied rather than performed and left + unacknowledged. An unbounded queue turns a long partition into an unbounded liability: the + machine keeps acting, nothing can be confirmed, and the operator finds out when the disk fills. + + ## The one thing that is always recordable + + A queue-full refusal is written to the chain and is **not** itself queued. If it were, a full + queue could not record that it was full, which is a deadlock dressed as a safety property. Its + entry in the chain is a local fact about this machine's state, and the far side learns it when + the backlog drains and the acknowledgement gap closes. + """ + + import Ecto.Query + + alias Trinity.Receipts.{QueueEntry, Receipt} + alias Trinity.Repo.Receipts, as: Repo + + @default_bound 10_000 + + @doc """ + The maximum number of pending entries a single chain scope may hold. + + `config :trinity, :receipts, queue_bound: n`. The default is deliberately large: the bound exists + to stop an unbounded liability, not to be reached in ordinary operation. + """ + @spec bound() :: pos_integer() + def bound do + Application.get_env(:trinity, :receipts, []) |> Keyword.get(:queue_bound, @default_bound) + end + + @doc "Pending entries for a scope, oldest first." + @spec pending(String.t(), pos_integer()) :: [QueueEntry.t()] + def pending(scope, limit \\ 100) do + Repo.all( + from q in QueueEntry, + where: q.chain_scope == ^scope and q.status == "pending", + order_by: [asc: q.seq], + limit: ^limit + ) + end + + @doc "How many entries a scope has pending." + @spec depth(String.t()) :: non_neg_integer() + def depth(scope) do + Repo.one( + from q in QueueEntry, + where: q.chain_scope == ^scope and q.status == "pending", + select: count(q.id) + ) || 0 + end + + @doc "Every scope with pending entries." + @spec scopes_with_pending() :: [String.t()] + def scopes_with_pending do + Repo.all( + from q in QueueEntry, + where: q.status == "pending", + distinct: true, + select: q.chain_scope + ) + end + + @doc "True when a scope is at or past the bound, so a forwardable receipt must be refused." + @spec full?(String.t()) :: boolean() + def full?(scope), do: depth(scope) >= bound() + + @doc """ + The changeset for queueing a receipt, for use inside the writer's transaction. + + Returned rather than inserted so that a receipt and its queue entry land together or not at all. + A receipt in the chain with no queue entry is a receipt the far side will never see and nothing + will ever notice; a queue entry with no receipt is a promise about a row that does not exist. + """ + @spec entry_changeset(Receipt.t()) :: Ecto.Changeset.t() + def entry_changeset(%Receipt{} = row) do + QueueEntry.changeset(%QueueEntry{}, %{ + chain_scope: row.chain_scope, + seq: row.seq, + receipt_hash: row.receipt_hash, + kind: row.kind, + envelope: row |> Receipt.to_export() |> JSON.encode!(), + status: "pending", + queued_at: DateTime.utc_now() + }) + end + + @doc """ + Marks an entry acknowledged, refusing an acknowledgement that would leave an older one pending. + + `{:error, {:out_of_order, first_pending_seq}}` when something older in the same scope has not been + acknowledged. The far side may well have received them out of order; what it may not do is tell + this side that a later receipt is safely recorded while an earlier one is not, because the gap + that leaves is invisible afterwards. + """ + @spec ack(String.t(), String.t()) :: :ok | {:error, term()} + def ack(scope, receipt_hash) do + case Repo.one( + from q in QueueEntry, + where: q.chain_scope == ^scope and q.receipt_hash == ^receipt_hash + ) do + nil -> + {:error, :not_queued} + + %QueueEntry{status: "acked"} -> + :ok + + %QueueEntry{} = entry -> + ack_if_oldest(scope, entry) + end + end + + defp ack_if_oldest(scope, entry) do + case oldest_pending_seq(scope) do + seq when seq == entry.seq -> mark_acked(entry) + older -> {:error, {:out_of_order, older}} + end + end + + defp mark_acked(entry) do + entry + |> QueueEntry.changeset(%{status: "acked", acked_at: DateTime.utc_now()}) + |> Repo.update() + |> case do + {:ok, _} -> :ok + {:error, cs} -> {:error, {:ack_failed, cs.errors}} + end + end + + @doc "Records a failed delivery attempt without acknowledging it." + @spec fail(QueueEntry.t(), term()) :: :ok + def fail(%QueueEntry{} = entry, reason) do + entry + |> QueueEntry.changeset(%{ + attempts: entry.attempts + 1, + last_error: reason |> inspect() |> String.slice(0, 200) + }) + |> Repo.update() + + :ok + end + + defp oldest_pending_seq(scope) do + Repo.one( + from q in QueueEntry, + where: q.chain_scope == ^scope and q.status == "pending", + select: min(q.seq) + ) + end +end diff --git a/lib/trinity/receipts/queue_entry.ex b/lib/trinity/receipts/queue_entry.ex new file mode 100644 index 0000000..c258a2a --- /dev/null +++ b/lib/trinity/receipts/queue_entry.ex @@ -0,0 +1,58 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Receipts.QueueEntry do + @moduledoc """ + One receipt awaiting acknowledgement by the authority adapter (slice 026). + + `envelope` is the exported receipt byte for byte and is never rebuilt at send time, so a retry + hands the far side exactly the bytes that were signed. Only `status`, `acked_at`, `attempts` and + `last_error` are ever written after insert; the envelope and the identity of the row are fixed. + """ + use Ecto.Schema + + import Ecto.Changeset + + @type t :: %__MODULE__{} + + @primary_key {:id, Trinity.UUID, autogenerate: true} + schema "receipt_queue" do + field :chain_scope, :string + field :seq, :integer + field :receipt_hash, :string + field :kind, :string + field :envelope, :string + field :status, :string, default: "pending" + field :attempts, :integer, default: 0 + field :last_error, :string + field :queued_at, :utc_datetime_usec + field :acked_at, :utc_datetime_usec + end + + @statuses ~w(pending acked) + + @doc "The two statuses an entry can hold." + @spec statuses() :: [String.t()] + def statuses, do: @statuses + + @doc false + @spec changeset(t(), map()) :: Ecto.Changeset.t() + def changeset(entry, attrs) do + entry + |> cast(attrs, [ + :chain_scope, + :seq, + :receipt_hash, + :kind, + :envelope, + :status, + :attempts, + :last_error, + :queued_at, + :acked_at + ]) + |> validate_required([:chain_scope, :seq, :receipt_hash, :kind, :envelope, :queued_at]) + |> validate_inclusion(:status, @statuses) + |> unique_constraint(:receipt_hash) + |> unique_constraint([:chain_scope, :seq]) + end +end diff --git a/lib/trinity/receipts/signer.ex b/lib/trinity/receipts/signer.ex index 2159e4a..9924602 100644 --- a/lib/trinity/receipts/signer.ex +++ b/lib/trinity/receipts/signer.ex @@ -18,7 +18,15 @@ defmodule Trinity.Receipts.Signer do @doc "The algorithm this implementation signs with." @callback algorithm() :: algorithm() - @doc "The scheme string a receipt of this family carries: `receipt_v2_`." + @doc """ + The scheme string a receipt of this family carries, at the current version: + `receipt__`. + + Slice 026 moved the current version from `v2` to `v3`, because `v3` carries a hybrid logical + clock in the signed payload and `v2` does not. Both are accepted on read (`accepted_schemes/0`): + a chain that spans the bump verifies row by row under each row's own scheme, which is what makes + this a scheme bump rather than an edit to bytes already signed. + """ @callback scheme() :: String.t() @doc "True where this runtime can sign and verify with this algorithm." @@ -56,12 +64,79 @@ defmodule Trinity.Receipts.Signer do def impl(algorithm) when is_map_key(@implementations, algorithm), do: Map.fetch!(@implementations, algorithm) - @doc "The implementation whose scheme string this is, or `:error`." + # Every scheme version this tree can read, oldest first. `v2` is slice 024's; `v3` is slice 026's + # and carries the clock. A version is removed from here only when no chain in the world still has + # a row of it, which is not a thing this project can know, so in practice it is never removed. + @scheme_versions ["v2", "v3"] + + # The version new rows are written under. + @current_scheme_version "v3" + + @doc "The scheme versions this tree can verify, oldest first." + @spec scheme_versions() :: [String.t()] + def scheme_versions, do: @scheme_versions + + @doc "The scheme version new rows are written under." + @spec current_scheme_version() :: String.t() + def current_scheme_version, do: @current_scheme_version + + @doc """ + Every scheme this tree accepts on read: each algorithm at each readable version. + + This is the verifier's default allow-list rather than the current schemes alone. Defaulting to + the current schemes would mean that bumping the version made every previously signed row fail to + verify, which is the opposite of what a bump is for. + """ + @spec accepted_schemes() :: [String.t()] + def accepted_schemes do + for v <- @scheme_versions, a <- algorithms(), do: "receipt_#{v}_#{a}" + end + + @doc """ + Splits a scheme string into its version and family, or `:error`. + + The shape is `receipt__`; no family contains an underscore, which is why + splitting into two parts is safe and is asserted by a test rather than assumed. + """ + @spec parse_scheme(String.t()) :: {:ok, String.t(), String.t()} | :error + def parse_scheme("receipt_" <> rest) do + case String.split(rest, "_", parts: 2) do + [version, family] when version in @scheme_versions -> {:ok, version, family} + _ -> :error + end + end + + def parse_scheme(_), do: :error + + @doc """ + True when a scheme's signed payload carries a hybrid logical clock. + + `v2` predates slice 026 and has no clock; asking a `v2` row for one is a question about a row + that was signed before the field existed, and the verifier must not treat its absence as a fault. + """ + @spec clocked?(String.t()) :: boolean() + def clocked?(scheme) do + case parse_scheme(scheme) do + {:ok, version, _family} -> version >= "v3" + :error -> false + end + end + + @doc "The implementation whose scheme string this is, at any readable version, or `:error`." @spec impl_for_scheme(String.t()) :: {:ok, module()} | :error def impl_for_scheme(scheme) do - case Enum.find(@implementations, fn {_, m} -> m.scheme() == scheme end) do - {_, m} -> {:ok, m} + with {:ok, _version, family} <- parse_scheme(scheme), + {:ok, algorithm} <- family_to_algorithm(family) do + {:ok, Map.fetch!(@implementations, algorithm)} + else + _ -> :error + end + end + + defp family_to_algorithm(family) do + case Enum.find(algorithms(), &(Atom.to_string(&1) == family)) do nil -> :error + algorithm -> {:ok, algorithm} end end diff --git a/lib/trinity/receipts/signer/ed25519.ex b/lib/trinity/receipts/signer/ed25519.ex index 7527517..3722024 100644 --- a/lib/trinity/receipts/signer/ed25519.ex +++ b/lib/trinity/receipts/signer/ed25519.ex @@ -8,7 +8,7 @@ defmodule Trinity.Receipts.Signer.Ed25519 do def algorithm, do: :ed25519 @impl true - def scheme, do: "receipt_v2_ed25519" + def scheme, do: "receipt_v3_ed25519" @impl true def available?, do: :ed25519 in :crypto.supports(:curves) and :crypto.info_fips() != :enabled diff --git a/lib/trinity/receipts/signer/mldsa87.ex b/lib/trinity/receipts/signer/mldsa87.ex index 84c9366..2403a49 100644 --- a/lib/trinity/receipts/signer/mldsa87.ex +++ b/lib/trinity/receipts/signer/mldsa87.ex @@ -18,7 +18,7 @@ defmodule Trinity.Receipts.Signer.MLDSA87 do def algorithm, do: :mldsa87 @impl true - def scheme, do: "receipt_v2_mldsa87" + def scheme, do: "receipt_v3_mldsa87" @impl true def available?, do: :mldsa87 in :crypto.supports(:public_keys) diff --git a/lib/trinity/receipts/signer/p384.ex b/lib/trinity/receipts/signer/p384.ex index 2796df9..c86156d 100644 --- a/lib/trinity/receipts/signer/p384.ex +++ b/lib/trinity/receipts/signer/p384.ex @@ -12,7 +12,7 @@ defmodule Trinity.Receipts.Signer.P384 do def algorithm, do: :p384 @impl true - def scheme, do: "receipt_v2_p384" + def scheme, do: "receipt_v3_p384" @impl true def available?, do: :secp384r1 in :crypto.supports(:curves) diff --git a/lib/trinity/receipts/supervisor.ex b/lib/trinity/receipts/supervisor.ex index fdbf7e8..4ba1b19 100644 --- a/lib/trinity/receipts/supervisor.ex +++ b/lib/trinity/receipts/supervisor.ex @@ -29,7 +29,12 @@ defmodule Trinity.Receipts.Supervisor do end children = [ - {DynamicSupervisor, name: Trinity.Receipts.WriterSupervisor, strategy: :one_for_one} + {DynamicSupervisor, name: Trinity.Receipts.WriterSupervisor, strategy: :one_for_one}, + # Slice 026: drains the outbound queue to the authority in force. Started even where no + # adapter exports `forward_receipt/2`: it then finds nothing to do, which is a configuration + # fact rather than a failure, and the local authority acknowledges immediately so that + # standalone Trinity exercises the same queue-and-acknowledge path an adapter would. + Trinity.Receipts.Forwarder ] Supervisor.init(children, strategy: :one_for_one) diff --git a/lib/trinity/receipts/verifier.ex b/lib/trinity/receipts/verifier.ex index 75587a3..748f4d9 100644 --- a/lib/trinity/receipts/verifier.ex +++ b/lib/trinity/receipts/verifier.ex @@ -49,7 +49,7 @@ defmodule Trinity.Receipts.Verifier do opts ) do schemes = - Keyword.get(opts, :schemes, Enum.map(Signer.algorithms(), &Signer.impl(&1).scheme())) + Keyword.get(opts, :schemes, Signer.accepted_schemes()) require_coverage = Keyword.get(opts, :require_coverage, true) diff --git a/priv/repo_receipts/migrations/20260926140000_create_receipt_queue.exs b/priv/repo_receipts/migrations/20260926140000_create_receipt_queue.exs new file mode 100644 index 0000000..05a53b1 --- /dev/null +++ b/priv/repo_receipts/migrations/20260926140000_create_receipt_queue.exs @@ -0,0 +1,43 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Repo.Receipts.Migrations.CreateReceiptQueue do + @moduledoc """ + Slice 026. The durable outbound queue: one row per receipt awaiting acknowledgement by the + authority adapter, per chain scope. + + `envelope` is the exported receipt **byte for byte**. The external authority plane's maintainers + answered the question this slice was blocked on by saying that store-and-forward does not change + their side as long as the acknowledgement carries the envelope they signed, unmodified: their + verifiers check the signature over those bytes offline, so a queue may wrap, delay or re-deliver + an envelope and may never re-sign it or alter a leaf. Storing the bytes rather than rebuilding + them at send time is what makes that true here rather than merely intended. + + Nothing in this table is ever rewritten except `status` and `acked_at`. The receipt itself lives + in `receipts` and is never touched by the queue. + """ + use Ecto.Migration + + def change do + create table(:receipt_queue, primary_key: false) do + add :id, :binary_id, primary_key: true + add :chain_scope, :string, null: false + add :seq, :integer, null: false + add :receipt_hash, :string, null: false + add :kind, :string, null: false + add :envelope, :text, null: false + add :status, :string, null: false, default: "pending" + add :attempts, :integer, null: false, default: 0 + add :last_error, :string + add :queued_at, :utc_datetime_usec, null: false + add :acked_at, :utc_datetime_usec + end + + # One queue entry per receipt: a receipt cannot be queued twice, so a re-delivery is a resend + # of the same row rather than a second row that an adapter would see as a second fact. + create unique_index(:receipt_queue, [:receipt_hash]) + create unique_index(:receipt_queue, [:chain_scope, :seq]) + + # The drain reads pending rows for a scope in seq order; the depth check counts them. + create index(:receipt_queue, [:chain_scope, :status, :seq]) + end +end diff --git a/test/fips/receipts_test.exs b/test/fips/receipts_test.exs index b77f3ec..c0c5e57 100644 --- a/test/fips/receipts_test.exs +++ b/test/fips/receipts_test.exs @@ -23,13 +23,13 @@ defmodule Trinity.Fips.ReceiptsTest do refute Signer.Ed25519.available?() assert Signer.P384.available?() assert {:ok, :p384} = KeyCustody.select() - assert %{algorithm: :p384, scheme: "receipt_v2_p384"} = KeyCustody.selected() + assert %{algorithm: :p384, scheme: "receipt_v3_p384"} = KeyCustody.selected() boot = Receipts.boot_receipt() assert boot.subject["signer"]["algorithm"] == "p384" - assert boot.subject["signer"]["scheme"] == "receipt_v2_p384" + assert boot.subject["signer"]["scheme"] == "receipt_v3_p384" assert boot.subject["fips"] == "enabled" - assert boot.scheme == "receipt_v2_p384" + assert boot.scheme == "receipt_v3_p384" end test "no effect is denied for want of a signer: an effect runs, its receipts are P-384 and verify" do @@ -51,13 +51,13 @@ defmodule Trinity.Fips.ReceiptsTest do rows = Receipts.list(scope) assert Enum.map(rows, & &1.kind) == ["decision", "effect", "effect"] - assert Enum.all?(rows, &(&1.scheme == "receipt_v2_p384" and is_binary(&1.signature))) + assert Enum.all?(rows, &(&1.scheme == "receipt_v3_p384" and is_binary(&1.signature))) :ok = Receipts.stop_writer(scope) {:ok, export} = Receipts.export(scope) assert {:ok, %{receipts: 3}} = Verifier.verify(export) # And an Ed25519-only verifier refuses the chain at the scheme string, not at a signature. - assert {:error, 1, {:scheme_not_allowed, 1, "receipt_v2_p384"}} = - Verifier.verify(export, schemes: ["receipt_v2_ed25519"]) + assert {:error, 1, {:scheme_not_allowed, 1, "receipt_v3_p384"}} = + Verifier.verify(export, schemes: ["receipt_v3_ed25519"]) end end diff --git a/test/support/authority/adapters.ex b/test/support/authority/adapters.ex index 416ea60..2e814d4 100644 --- a/test/support/authority/adapters.ex +++ b/test/support/authority/adapters.ex @@ -22,3 +22,81 @@ defmodule Trinity.TestAuthority.Full do @impl true def receipt(_kind, _attrs), do: {:ok, :recorded_elsewhere} end + +defmodule Trinity.TestAuthority.Unreachable do + @moduledoc """ + Slice 026: an adapter whose link is down. Every forward fails; nothing is ever acknowledged. + + A partition is the case store-and-forward exists for, so this is the ordinary condition under + test rather than an error injected into it. + """ + @behaviour Trinity.Authority + def stage(staged, _ctx), do: {:ok, staged} + def decide(_staged, decision, _ctx), do: {:ok, decision, "test"} + def execute(_staged, _decision, _ctx), do: {:error, :not_used} + def receipt(_kind, _attrs), do: {:error, :not_used} + def forward_receipt(_envelope, _meta), do: {:error, :unreachable} +end + +defmodule Trinity.TestAuthority.Recorder do + @moduledoc """ + Slice 026: an adapter that accepts every forward and records the order it was offered them. + + The order is the point: a far side holding row 4 but not row 3 has a gap it cannot see, so the + sequence this records is what proves the queue drains oldest first. + """ + @behaviour Trinity.Authority + def stage(staged, _ctx), do: {:ok, staged} + def decide(_staged, decision, _ctx), do: {:ok, decision, "test"} + def execute(_staged, _decision, _ctx), do: {:error, :not_used} + def receipt(_kind, _attrs), do: {:error, :not_used} + + @doc "Clears the record before a test uses it." + def start, do: :persistent_term.put({__MODULE__, :seen}, []) + + @doc "The seqs this adapter was offered, in the order it was offered them." + def seen, do: :persistent_term.get({__MODULE__, :seen}, []) |> Enum.reverse() + + def forward_receipt(_envelope, meta) do + :persistent_term.put( + {__MODULE__, :seen}, + [meta["seq"] | :persistent_term.get({__MODULE__, :seen}, [])] + ) + + :ok + end +end + +defmodule Trinity.TestAuthority.FailsFrom do + @moduledoc """ + Slice 026: an adapter whose link drops partway through a drain. + + It accepts every forward below a seq set with `start/1` and refuses from there on, so a test can + assert that a scope stops at its first refusal instead of skipping past it. + """ + @behaviour Trinity.Authority + def stage(staged, _ctx), do: {:ok, staged} + def decide(_staged, decision, _ctx), do: {:ok, decision, "test"} + def execute(_staged, _decision, _ctx), do: {:error, :not_used} + def receipt(_kind, _attrs), do: {:error, :not_used} + + @doc "Refuse from this seq onward, and clear the record." + def start(from) do + :persistent_term.put({__MODULE__, :from}, from) + :persistent_term.put({__MODULE__, :seen}, []) + end + + @doc "The seqs this adapter was offered, in order." + def seen, do: :persistent_term.get({__MODULE__, :seen}, []) |> Enum.reverse() + + def forward_receipt(_envelope, meta) do + :persistent_term.put( + {__MODULE__, :seen}, + [meta["seq"] | :persistent_term.get({__MODULE__, :seen}, [])] + ) + + if meta["seq"] >= :persistent_term.get({__MODULE__, :from}, 0), + do: {:error, :link_down}, + else: :ok + end +end diff --git a/test/trinity/receipts/chain_writer_test.exs b/test/trinity/receipts/chain_writer_test.exs index 03e844c..c364440 100644 --- a/test/trinity/receipts/chain_writer_test.exs +++ b/test/trinity/receipts/chain_writer_test.exs @@ -73,7 +73,7 @@ defmodule Trinity.Receipts.ChainWriterTest do body = JSON.decode!(hd(rows).signed_payload) assert Map.keys(body) |> Enum.sort() == - ~w(at chain_scope decision fingerprint key_id kind prev_hash scheme seq subject) + ~w(at chain_scope clock decision fingerprint key_id kind prev_hash scheme seq subject) assert body["scheme"] == impl.scheme() end @@ -268,29 +268,54 @@ defmodule Trinity.Receipts.ChainWriterTest do files = out |> String.split("\n", trim: true) assert length(files) > 50 - writes = ~w(insert insert! insert_all update update_all delete delete_all) - - # A file writes through the receipts Repo when it names the module, or an alias of it, - # followed by a write function. Reads (`all`, `one`, `aggregate`) do not count. - inserters = - for f <- files, - src = File.read!(f), - aliases = - Regex.scan(~r/alias Trinity\.Repo\.Receipts(?:, as: ([A-Z]\w*))?/, src) - |> Enum.map(fn - [_, as] -> as - [_] -> "Receipts" - end), - names = ["Trinity.Repo.Receipts" | aliases], - Enum.any?(names, fn n -> - Enum.any?(writes, &String.contains?(src, n <> "." <> &1 <> "(")) - end), - do: f + # Slice 026 split this in two. The property slice 024 cared about is that nothing but the + # writer brings a chain row into existence or removes one, and that is stated here directly. + # The outbound queue (slice 026) writes through the same Repo, to `receipt_queue`, and only + # ever updates a row it did not create; lumping it in with the inserters would have meant + # either a false failure or widening the census until it no longer said anything. + creates = ~w(insert insert! insert_all delete delete_all) + mutates = creates ++ ~w(update update_all) + + callers = fn src, funs -> + aliases = + Regex.scan(~r/alias Trinity\.Repo\.Receipts(?:, as: ([A-Z]\w*))?/, src) + |> Enum.map(fn + [_, as] -> as + [_] -> "Receipts" + end) + + names = ["Trinity.Repo.Receipts" | aliases] + + Enum.any?(names, fn n -> + Enum.any?(funs, &String.contains?(src, n <> "." <> &1 <> "(")) + end) + end + + inserters = for f <- files, callers.(File.read!(f), creates), do: f + writers = for f <- files, callers.(File.read!(f), mutates), do: f assert Enum.sort(inserters) == [ "lib/trinity/receipts/chain_writer.ex", "test/support/receipts/bypass_inserter.ex" - ] + ], + "something other than the ChainWriter creates or removes rows through the receipts Repo" + + assert Enum.sort(writers) == [ + "lib/trinity/receipts/chain_writer.ex", + "lib/trinity/receipts/queue.ex", + "test/support/receipts/bypass_inserter.ex" + ], + "a new writer appeared through the receipts Repo" + + # And the queue writes its own table only. It never names the chain's schemas in a write. + queue_src = File.read!("lib/trinity/receipts/queue.ex") + + for fun <- mutates do + refute queue_src =~ ~r/Repo\.#{fun}\(\s*%(Receipt|Checkpoint)\{/, + "the queue writes a chain schema through #{fun}" + end + + assert queue_src =~ "QueueEntry" assert Trinity.TestReceipts.BypassInserter in (:application.get_key(:trinity, :modules) |> elem(1)) diff --git a/test/trinity/receipts/clock_chain_test.exs b/test/trinity/receipts/clock_chain_test.exs new file mode 100644 index 0000000..d563180 --- /dev/null +++ b/test/trinity/receipts/clock_chain_test.exs @@ -0,0 +1,269 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Receipts.ClockChainTest do + @moduledoc """ + Slice 026: the clock in a real chain, and the scheme bump that carries it. + + The clock is in the **signed** payload, not in `meta`. That is the decision this file exists to + hold in place: a merge that orders two chains by an unsigned field orders them by something any + writer could rewrite afterwards, which would make the merge evidence of nothing. The cost is a + scheme bump, and the bump has to leave every row signed under the old scheme verifiable, which is + the other half of what is asserted here. + """ + use Trinity.DataCase, async: false + + alias Trinity.Receipts + alias Trinity.Receipts.{Clock, Envelope, KeyCustody, Receipt, Signer, Verifier} + + setup do + scope = "test:" <> Trinity.UUID.generate() + on_exit(fn -> Receipts.stop_writer(scope) end) + {:ok, scope: scope} + end + + defp decision(n), + do: %{ + kind: "decision", + subject: %{"n" => n}, + decision: %{"outcome" => "allow"}, + fingerprint: "ab" <> Integer.to_string(n) + } + + # `stop_writer/1` returns before the process is gone, so an append straight after it can reach a + # dying pid through the registry and exit with :noproc. `chain_writer_test.exs` already waits the + # same way; this test did not, and failed 2 runs in 3 until it did. + defp await_stopped(scope) do + if Trinity.Receipts.ChainWriter.whereis(scope) do + Process.sleep(10) + await_stopped(scope) + else + :ok + end + end + + defp clock_of(%Receipt{signed_payload: payload}) do + {:ok, body} = JSON.decode(payload) + {:ok, clock} = Clock.from_map(body["clock"]) + clock + end + + describe "AC1 (the clock half): every receipt carries one, and it only goes forward" do + test "every row of a chain carries a clock in its signed payload", %{scope: scope} do + for n <- 1..5, do: {:ok, _} = Receipts.append(scope, decision(n)) + + for row <- Receipts.list(scope) do + {:ok, body} = JSON.decode(row.signed_payload) + + assert {:ok, %Clock{}} = Clock.from_map(body["clock"]), + "row #{row.seq} carries no clock: #{inspect(body["clock"])}" + end + end + + test "the clocks are strictly increasing along the chain", %{scope: scope} do + for n <- 1..8, do: {:ok, _} = Receipts.append(scope, decision(n)) + + clocks = scope |> Receipts.list() |> Enum.map(&clock_of/1) + + clocks + |> Enum.zip(tl(clocks)) + |> Enum.each(fn {a, b} -> + assert Clock.compare(b, a) == :gt, + "clock did not advance between rows: #{inspect(a)} then #{inspect(b)}" + end) + end + + test "every row names this device", %{scope: scope} do + {:ok, row} = Receipts.append(scope, decision(1)) + assert clock_of(row).node == Clock.node_id() + end + + test "a chain survives a writer restart with its clock intact", %{scope: scope} do + for n <- 1..3, do: {:ok, _} = Receipts.append(scope, decision(n)) + before = scope |> Receipts.list() |> List.last() |> clock_of() + + :ok = Receipts.stop_writer(scope) + :ok = await_stopped(scope) + {:ok, _} = Receipts.append(scope, decision(4)) + + after_restart = scope |> Receipts.list() |> List.last() |> clock_of() + + assert Clock.compare(after_restart, before) == :gt, + "the writer rehydrated without reading the tail's clock, so the chain's clock " <> + "restarted and a row now appears to precede one already written" + end + end + + describe "the clock is signed, not decoration" do + test "editing the clock in a stored payload breaks verification", %{scope: scope} do + for n <- 1..3, do: {:ok, _} = Receipts.append(scope, decision(n)) + :ok = Receipts.stop_writer(scope) + {:ok, export} = Receipts.export(scope) + + assert {:ok, _} = Verifier.verify(export) + + tampered = + update_in(export["receipts"], fn rows -> + Enum.map(rows, fn r -> + {:ok, body} = JSON.decode(r["signed_payload"]) + + if body["seq"] == 2 do + moved = %{body | "clock" => %{body["clock"] | "wall" => 0}} + %{r | "signed_payload" => Jcs.encode(moved)} + else + r + end + end) + end) + + assert {:error, _, _} = Verifier.verify(tampered), + "the clock was moved and the chain still verified. If the clock can be edited " <> + "after the fact then ordering two chains by it proves nothing." + end + end + + describe "AC3: a clock that does not follow the tail is refused, and the refusal is receipted" do + test "the offered clock is behind the chain tail", %{scope: scope} do + for n <- 1..3, do: {:ok, _} = Receipts.append(scope, decision(n)) + tail_clock = scope |> Receipts.list() |> List.last() |> clock_of() + + behind = %Clock{tail_clock | wall: tail_clock.wall - 1_000} + + assert {:error, {:clock_regression, _offered, _held}} = + Receipts.append(scope, Map.put(decision(99), :clock, behind)) + end + + test "the refusal is itself a receipt on the chain, naming both clocks", %{scope: scope} do + for n <- 1..3, do: {:ok, _} = Receipts.append(scope, decision(n)) + tail_clock = scope |> Receipts.list() |> List.last() |> clock_of() + behind = %Clock{tail_clock | wall: tail_clock.wall - 1_000} + + {:error, _} = Receipts.append(scope, Map.put(decision(99), :clock, behind)) + + refusal = + scope + |> Receipts.list() + |> Enum.find( + &(&1.subject_ref && String.starts_with?(&1.subject_ref, "clock_regression:")) + ) + + assert refusal, + "a receipt was refused and nothing recorded it. A refusal that leaves no record is " <> + "indistinguishable from the call never having been made." + + {:ok, body} = JSON.decode(refusal.signed_payload) + assert get_in(body, ["decision", "outcome"]) == "refused" + assert get_in(body, ["decision", "basis"]) == "clock" + assert get_in(body, ["subject", "offered_clock", "wall"]) == behind.wall + assert get_in(body, ["subject", "held_clock", "wall"]) == tail_clock.wall + end + + test "the refused row is not in the chain, and the chain still verifies", %{scope: scope} do + for n <- 1..3, do: {:ok, _} = Receipts.append(scope, decision(n)) + tail_clock = scope |> Receipts.list() |> List.last() |> clock_of() + behind = %Clock{tail_clock | wall: tail_clock.wall - 1_000} + + {:error, _} = Receipts.append(scope, Map.put(decision(99), :clock, behind)) + + refute Enum.any?(Receipts.list(scope), fn r -> + {:ok, body} = JSON.decode(r.signed_payload) + body["fingerprint"] == "ab99" + end), + "the refused receipt was written anyway" + + :ok = Receipts.stop_writer(scope) + {:ok, export} = Receipts.export(scope) + assert {:ok, _} = Verifier.verify(export) + end + + test "an equal clock is refused too: a row must follow, not tie", %{scope: scope} do + {:ok, _} = Receipts.append(scope, decision(1)) + tail_clock = scope |> Receipts.list() |> List.last() |> clock_of() + + assert {:error, {:clock_regression, _, _}} = + Receipts.append(scope, Map.put(decision(2), :clock, tail_clock)) + end + end + + describe "the scheme bump is a bump, not an edit" do + test "new rows are written under the current version", %{scope: scope} do + {:ok, row} = Receipts.append(scope, decision(1)) + assert {:ok, version, _family} = Signer.parse_scheme(row.scheme) + assert version == Signer.current_scheme_version() + assert Signer.clocked?(row.scheme) + end + + test "both versions resolve to the same signer family" do + for a <- Signer.algorithms() do + v2 = "receipt_v2_#{a}" + v3 = "receipt_v3_#{a}" + assert Signer.impl_for_scheme(v2) == Signer.impl_for_scheme(v3) + end + end + + test "v2 is readable and carries no clock; v3 carries one" do + refute Signer.clocked?("receipt_v2_ed25519") + assert Signer.clocked?("receipt_v3_ed25519") + assert "receipt_v2_ed25519" in Signer.accepted_schemes() + assert "receipt_v3_ed25519" in Signer.accepted_schemes() + end + + test "no signer family contains an underscore, which is what makes parsing the scheme safe" do + for a <- Signer.algorithms() do + refute Atom.to_string(a) =~ "_", + "#{a} contains an underscore, so receipt__ can no longer be " <> + "split into two parts and Signer.parse_scheme/1 is wrong" + end + end + + test "a row signed under v2 still verifies after the bump", %{scope: scope} do + # Built by hand because the writer only ever writes the current version. This is the whole + # claim of a scheme bump: bytes signed last week still verify this week. + %{scheme: current, key_id: key_id} = KeyCustody.selected() + {:ok, _version, family} = Signer.parse_scheme(current) + v2_scheme = "receipt_v2_#{family}" + + body = %{ + "scheme" => v2_scheme, + "seq" => 1, + "chain_scope" => scope, + "prev_hash" => nil, + "kind" => "decision", + "subject" => %{"n" => 1}, + "decision" => %{"outcome" => "allow"}, + "fingerprint" => "ab01", + "at" => DateTime.utc_now() |> DateTime.to_iso8601(), + "key_id" => key_id + } + + payload = Envelope.canonical(body) + bytes = Envelope.pae(Envelope.receipt_type(v2_scheme), payload) + {:ok, signature} = KeyCustody.sign(bytes) + + {:ok, registry} = Trinity.Receipts.KeyRegistry.read(KeyCustody.keys_dir()) + + export = %{ + "format" => "trinity-receipts-export/1", + "chain_scope" => scope, + "receipts" => [ + %{ + "seq" => 1, + "prev_hash" => nil, + "receipt_hash" => Envelope.hash(bytes), + "scheme" => v2_scheme, + "kind" => "decision", + "chain_scope" => scope, + "key_id" => key_id, + "signed_payload" => payload, + "signature_b64" => Base.encode64(signature) + } + ], + "checkpoints" => [], + "registry" => registry + } + + assert {:ok, %{receipts: 1}} = Verifier.verify(export), + "a row signed under the previous scheme no longer verifies. That makes the version " <> + "change an edit to history rather than a bump." + end + end +end diff --git a/test/trinity/receipts/clock_test.exs b/test/trinity/receipts/clock_test.exs new file mode 100644 index 0000000..cdd1005 --- /dev/null +++ b/test/trinity/receipts/clock_test.exs @@ -0,0 +1,130 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Receipts.ClockTest do + @moduledoc """ + Slice 026: the hybrid logical clock is monotone even when the host's wall clock is not. + + The property that matters is the one a partition depends on: a chain's clocks never go backwards, + whatever the machine underneath does. Everything else the clock offers is an assertion about an + untrusted reading, and the moduledoc says so. + """ + use ExUnit.Case, async: true + use ExUnitProperties + + alias Trinity.Receipts.Clock + + test "the first clock of a chain takes the wall reading" do + c = Clock.next(nil, 1_000) + assert c.wall == 1_000 + assert c.counter == 0 + assert c.node == Clock.node_id() + end + + test "a wall reading that moved forward resets the counter" do + prev = %Clock{wall: 1_000, counter: 7, node: Clock.node_id()} + assert %Clock{wall: 2_000, counter: 0} = Clock.next(prev, 2_000) + end + + test "a wall reading that stood still increments the counter" do + prev = %Clock{wall: 1_000, counter: 0, node: Clock.node_id()} + assert %Clock{wall: 1_000, counter: 1} = Clock.next(prev, 1_000) + end + + test "a wall clock that went BACKWARDS still produces a greater clock" do + # The case the whole design exists for. A host whose clock steps back must not be able to + # produce a receipt that appears to precede one it already wrote. + prev = %Clock{wall: 5_000, counter: 0, node: Clock.node_id()} + next = Clock.next(prev, 1_000) + + assert next.wall == 5_000, "the clock followed the wall backwards" + assert next.counter == 1 + assert Clock.compare(next, prev) == :gt + end + + property "next/2 is strictly greater than its predecessor, for any sequence of wall readings" do + check all(walls <- list_of(integer(0..10_000), min_length: 1, max_length: 40)) do + {_last, pairs} = + Enum.reduce(walls, {nil, []}, fn wall, {prev, acc} -> + next = Clock.next(prev, wall) + {next, [{prev, next} | acc]} + end) + + for {prev, next} <- pairs, prev != nil do + assert Clock.compare(next, prev) == :gt, + "clock went backwards or stalled: #{inspect(prev)} then #{inspect(next)}" + end + end + end + + describe "compare/2" do + test "orders by wall, then counter, then node" do + a = %Clock{wall: 1, counter: 0, node: "a"} + assert Clock.compare(a, %Clock{wall: 2, counter: 0, node: "a"}) == :lt + assert Clock.compare(a, %Clock{wall: 1, counter: 1, node: "a"}) == :lt + assert Clock.compare(a, %Clock{wall: 1, counter: 0, node: "b"}) == :lt + assert Clock.compare(a, a) == :eq + end + + property "it is a total order: exactly one of lt, eq, gt, and it is antisymmetric" do + gen = fn -> {integer(0..100), integer(0..10), member_of(["a", "b", "c"])} end + {w, c, n} = gen.() + + check all( + a <- map({w, c, n}, fn {w, c, n} -> %Clock{wall: w, counter: c, node: n} end), + b <- map({w, c, n}, fn {w, c, n} -> %Clock{wall: w, counter: c, node: n} end) + ) do + case Clock.compare(a, b) do + :eq -> assert Clock.compare(b, a) == :eq + :lt -> assert Clock.compare(b, a) == :gt + :gt -> assert Clock.compare(b, a) == :lt + end + end + end + end + + describe "concurrent?/2" do + test "same device is never concurrent with itself" do + a = %Clock{wall: 1, counter: 0, node: "a"} + refute Clock.concurrent?(a, a) + end + + test "different devices at the same reading are concurrent" do + assert Clock.concurrent?( + %Clock{wall: 1, counter: 0, node: "a"}, + %Clock{wall: 1, counter: 0, node: "b"} + ) + end + + test "different devices at different readings are ordered, not concurrent" do + refute Clock.concurrent?( + %Clock{wall: 1, counter: 0, node: "a"}, + %Clock{wall: 2, counter: 0, node: "b"} + ) + end + end + + describe "the signed form" do + test "round-trips through the payload shape" do + c = %Clock{wall: 123, counter: 4, node: "dev"} + assert {:ok, ^c} = c |> Clock.to_map() |> Clock.from_map() + end + + test "to_map/1 carries string keys and nothing else, so it canonicalises" do + m = Clock.to_map(%Clock{wall: 1, counter: 2, node: "d"}) + assert Map.keys(m) |> Enum.sort() == ["counter", "node", "wall"] + assert is_binary(Jcs.encode(m)) + end + + test "from_map/1 refuses a shape that is not a clock" do + assert Clock.from_map(%{"wall" => -1, "counter" => 0, "node" => "d"}) == :error + assert Clock.from_map(%{"wall" => 1, "counter" => 0}) == :error + assert Clock.from_map(%{"wall" => "1", "counter" => 0, "node" => "d"}) == :error + assert Clock.from_map(nil) == :error + end + end + + test "the device id is stable across calls" do + assert Clock.node_id() == Clock.node_id() + assert byte_size(Clock.node_id()) > 8 + end +end diff --git a/test/trinity/receipts/merge_test.exs b/test/trinity/receipts/merge_test.exs new file mode 100644 index 0000000..8105b74 --- /dev/null +++ b/test/trinity/receipts/merge_test.exs @@ -0,0 +1,223 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Receipts.MergeTest do + @moduledoc """ + Slice 026, AC2: two chains that diverged during a partition merge on reconnect with neither + rewritten and a conflict receipt for each divergence. + + The thing being asserted is mostly a negative. A merge may not make two accounts agree, because + nothing available after a partition can turn two true accounts of what each device did into one + account of what happened. What it may do is say precisely where they part company, and sign that. + """ + use Trinity.DataCase, async: false + use ExUnitProperties + + alias Trinity.Receipts + alias Trinity.Receipts.{Merge, Merkle, Verifier} + + setup do + scope = "test:" <> Trinity.UUID.generate() + on_exit(fn -> Receipts.stop_writer(scope) end) + {:ok, scope: scope} + end + + defp decision(n), + do: %{ + kind: "decision", + subject: %{"n" => n}, + decision: %{"outcome" => "allow"}, + fingerprint: "ab" <> Integer.to_string(n) + } + + # A second device's chain, built by running a second scope and relabelling its export to the + # first scope's name. Two devices genuinely holding the same scope is exactly the partition case, + # and this is the only way to produce one inside a single test VM. + defp remote_chain(n, opts \\ []) do + other = "test:" <> Trinity.UUID.generate() + from = Keyword.get(opts, :from, 1) + + for i <- from..(from + n - 1), do: {:ok, _} = Receipts.append(other, decision(i)) + + :ok = Receipts.stop_writer(other) + {:ok, export} = Receipts.export(other) + export + end + + defp local_hashes(scope), do: scope |> Receipts.list() |> Enum.map(& &1.receipt_hash) + + describe "Merkle heads" do + test "an empty chain has RFC 6962's empty-tree head" do + assert Merkle.root([]) == :crypto.hash(:sha256, "") |> Base.encode16(case: :lower) + end + + test "the head changes when any row changes" do + assert Merkle.root(["a", "b", "c"]) != Merkle.root(["a", "b", "d"]) + end + + test "order matters: the same rows in a different order are a different chain" do + assert Merkle.root(["a", "b"]) != Merkle.root(["b", "a"]) + end + + test "a leaf is not an interior node, which is what stops the second-preimage attack" do + # RFC 6962 section 2.1. Without the distinct prefixes, an interior node's hash could be + # presented as a leaf and a shorter tree forged to match a longer one's head. + left = Merkle.leaf("a") + right = Merkle.leaf("b") + interior = Merkle.node_hash(left, right) + + assert Merkle.leaf(left <> right) != interior + end + + test "an odd node is promoted, not duplicated" do + # Duplicating it is the Bitcoin construction (CVE-2012-2459) and lets distinct leaf sets + # share a root, which here would mean two different chains comparing as identical. + assert Merkle.root(["a", "b", "c"]) != Merkle.root(["a", "b", "c", "c"]) + end + + property "two lists share a head only when they are equal" do + check all( + a <- list_of(string(:alphanumeric, min_length: 1), max_length: 8), + b <- list_of(string(:alphanumeric, min_length: 1), max_length: 8) + ) do + if Merkle.root(a) == Merkle.root(b), do: assert(a == b) + end + end + end + + describe "AC2: chains that did not diverge" do + test "identical chains report identical and write nothing", %{scope: scope} do + for n <- 1..4, do: {:ok, _} = Receipts.append(scope, decision(n)) + :ok = Receipts.stop_writer(scope) + {:ok, export} = Receipts.export(scope) + + before = Receipts.list(scope) |> length() + + assert {:ok, :identical, summary} = Merge.merge(scope, export) + assert summary.local_head == summary.remote_head + assert summary.conflicts == [] + + assert Receipts.list(scope) |> length() == before, + "a merge of two identical chains wrote a receipt. That turns the chain into a log " <> + "of its own housekeeping." + end + + test "head/1 is the same value the summary reports", %{scope: scope} do + for n <- 1..3, do: {:ok, _} = Receipts.append(scope, decision(n)) + assert Merge.head(scope) == Merkle.root(local_hashes(scope)) + end + end + + describe "AC2: chains that diverged" do + test "a conflict receipt is written for each differing position", %{scope: scope} do + for n <- 1..3, do: {:ok, _} = Receipts.append(scope, decision(n)) + remote = remote_chain(3) + + assert {:ok, :diverged, summary} = Merge.merge(scope, remote) + + assert summary.local_head != summary.remote_head + assert summary.conflicts == [1, 2, 3] + + conflicts = + scope + |> Receipts.list() + |> Enum.filter( + &(&1.subject_ref && String.starts_with?(&1.subject_ref, "merge_conflict:")) + ) + + assert length(conflicts) == 3 + + for row <- conflicts do + {:ok, body} = JSON.decode(row.signed_payload) + assert get_in(body, ["decision", "outcome"]) == "conflict" + assert get_in(body, ["decision", "basis"]) == "merge" + assert get_in(body, ["subject", "local_receipt_hash"]) + assert get_in(body, ["subject", "remote_receipt_hash"]) + assert get_in(body, ["subject", "remote_device"]) + end + end + + test "neither side is rewritten", %{scope: scope} do + for n <- 1..3, do: {:ok, _} = Receipts.append(scope, decision(n)) + before = local_hashes(scope) + + remote = remote_chain(3) + remote_before = remote["receipts"] |> Enum.map(& &1["receipt_hash"]) + + {:ok, :diverged, _} = Merge.merge(scope, remote) + + after_merge = local_hashes(scope) + + assert Enum.take(after_merge, length(before)) == before, + "the merge altered rows that were already in the local chain" + + assert remote["receipts"] |> Enum.map(& &1["receipt_hash"]) == remote_before, + "the merge altered the remote export it was handed" + end + + test "the local chain still verifies after the merge", %{scope: scope} do + for n <- 1..3, do: {:ok, _} = Receipts.append(scope, decision(n)) + {:ok, :diverged, _} = Merge.merge(scope, remote_chain(3)) + + :ok = Receipts.stop_writer(scope) + {:ok, export} = Receipts.export(scope) + assert {:ok, _} = Verifier.verify(export) + end + + test "a remote longer than the local reports the positions only it has", %{scope: scope} do + for n <- 1..2, do: {:ok, _} = Receipts.append(scope, decision(n)) + + assert {:ok, :diverged, summary} = Merge.merge(scope, remote_chain(4)) + assert summary.conflicts == [1, 2, 3, 4] + + row = + scope + |> Receipts.list() + |> Enum.find(&(&1.subject_ref == "merge_conflict:#{scope}:4")) + + {:ok, body} = JSON.decode(row.signed_payload) + + assert get_in(body, ["decision", "reason"]) =~ "the remote chain has a row", + "a position the remote has and this chain does not was not described as such" + end + + test "a conflict names both heads, so a reader can check the comparison", %{scope: scope} do + {:ok, _} = Receipts.append(scope, decision(1)) + remote = remote_chain(1) + + {:ok, :diverged, summary} = Merge.merge(scope, remote) + + row = + scope + |> Receipts.list() + |> Enum.find(&(&1.subject_ref && String.starts_with?(&1.subject_ref, "merge_conflict:"))) + + {:ok, body} = JSON.decode(row.signed_payload) + + assert get_in(body, ["subject", "local_head"]) == summary.local_head + assert get_in(body, ["subject", "remote_head"]) == summary.remote_head + end + end + + describe "what a merge refuses" do + test "a remote export that does not verify is refused before anything is compared", %{ + scope: scope + } do + {:ok, _} = Receipts.append(scope, decision(1)) + remote = remote_chain(2) + + tampered = + update_in(remote["receipts"], fn rows -> + Enum.map(rows, fn r -> %{r | "receipt_hash" => String.duplicate("0", 64)} end) + end) + + assert {:error, {:remote_does_not_verify, _, _}} = Merge.merge(scope, tampered) + + refute Enum.any?( + Receipts.list(scope), + &(&1.subject_ref && String.starts_with?(&1.subject_ref, "merge_conflict:")) + ), + "a conflict was recorded against a chain that does not verify. That lets anyone who " <> + "can reach the merge write conflict receipts by handing it invented rows." + end + end +end diff --git a/test/trinity/receipts/queue_test.exs b/test/trinity/receipts/queue_test.exs new file mode 100644 index 0000000..9cccd36 --- /dev/null +++ b/test/trinity/receipts/queue_test.exs @@ -0,0 +1,246 @@ +# SPDX-FileCopyrightText: Sudo Apt Holdings LLC +# SPDX-License-Identifier: Apache-2.0 +defmodule Trinity.Receipts.QueueTest do + @moduledoc """ + Slice 026, AC1 and AC4: the outbound queue under a partition, and its bound. + + A partition is the case this exists for rather than an exception to it. With the adapter + unreachable the machine keeps working, the queue grows, and nothing is acknowledged; when the + link returns, every queued receipt is acknowledged in order. Past the bound the machine stops + accepting effects instead of performing them unacknowledged. + """ + use Trinity.DataCase, async: false + + alias Trinity.Receipts + alias Trinity.Receipts.{Forwarder, Queue, QueueEntry} + alias Trinity.Repo.Receipts, as: RRepo + alias Trinity.TestAuthority + + setup do + scope = "test:" <> Trinity.UUID.generate() + on_exit(fn -> Receipts.stop_writer(scope) end) + {:ok, scope: scope} + end + + defp decision(n), + do: %{ + kind: "decision", + subject: %{"n" => n}, + decision: %{"outcome" => "allow"}, + fingerprint: "ab" <> Integer.to_string(n) + } + + # The same route `test/trinity/mcp/auth/embedded_test.exs` already uses: the selection is a + # persistent_term written at boot, and a test that wants a different authority writes it back. + @selection_key {Trinity.Authority.Selection, :selected} + + defp with_authority(module, fun) do + old = :persistent_term.get(@selection_key, Trinity.Authority.Local) + :persistent_term.put(@selection_key, module) + + try do + fun.() + after + :persistent_term.put(@selection_key, old) + end + end + + defp with_bound(n, fun) do + old = Application.get_env(:trinity, :receipts, []) + Application.put_env(:trinity, :receipts, Keyword.put(old, :queue_bound, n)) + + try do + fun.() + after + Application.put_env(:trinity, :receipts, old) + end + end + + describe "AC1: with the adapter unreachable the queue grows; with it back, all are acked in order" do + test "every receipt written is queued", %{scope: scope} do + for n <- 1..5, do: {:ok, _} = Receipts.append(scope, decision(n)) + + assert Queue.depth(scope) == 5 + + entries = Queue.pending(scope, 100) + assert Enum.map(entries, & &1.seq) == [1, 2, 3, 4, 5] + + hashes = scope |> Receipts.list() |> Enum.map(& &1.receipt_hash) + assert Enum.map(entries, & &1.receipt_hash) == hashes + end + + test "the queued envelope is the exported receipt byte for byte", %{scope: scope} do + {:ok, row} = Receipts.append(scope, decision(1)) + [entry] = Queue.pending(scope, 10) + + assert entry.envelope == row |> Trinity.Receipts.Receipt.to_export() |> JSON.encode!(), + "the envelope was rebuilt rather than stored. The far side verifies a signature over " <> + "these bytes offline, so a queue that reconstructs them can break verification " <> + "without anything here failing." + end + + test "effects proceed and nothing is acknowledged while the adapter is unreachable", %{ + scope: scope + } do + with_authority(TestAuthority.Unreachable, fn -> + for n <- 1..4, do: {:ok, _} = Receipts.append(scope, decision(n)) + + assert %{acked: 0, failed: failed} = Forwarder.drain(scope) + assert failed > 0 + + assert Queue.depth(scope) == 4, "an unreachable adapter acknowledged something" + end) + end + + test "a failed offer records the attempt and the reason, and drops nothing", %{scope: scope} do + with_authority(TestAuthority.Unreachable, fn -> + {:ok, _} = Receipts.append(scope, decision(1)) + Forwarder.drain(scope) + + [entry] = Queue.pending(scope, 10) + assert entry.attempts == 1 + assert entry.last_error =~ "unreachable" + assert entry.status == "pending" + end) + end + + test "when the adapter is back, every queued receipt is acknowledged, oldest first", %{ + scope: scope + } do + with_authority(TestAuthority.Unreachable, fn -> + for n <- 1..6, do: {:ok, _} = Receipts.append(scope, decision(n)) + Forwarder.drain(scope) + assert Queue.depth(scope) == 6 + end) + + with_authority(TestAuthority.Recorder, fn -> + TestAuthority.Recorder.start() + + assert %{acked: 6, failed: 0} = Forwarder.drain(scope) + assert Queue.depth(scope) == 0 + + assert TestAuthority.Recorder.seen() == [1, 2, 3, 4, 5, 6], + "the adapter was offered receipts out of order. A far side holding row 4 but not " <> + "row 3 has a gap it cannot see." + end) + end + + test "a scope stops at its first refusal rather than skipping past it", %{scope: scope} do + for n <- 1..5, do: {:ok, _} = Receipts.append(scope, decision(n)) + + with_authority(TestAuthority.FailsFrom, fn -> + TestAuthority.FailsFrom.start(3) + assert %{acked: 2, failed: 1} = Forwarder.drain(scope) + + assert Queue.depth(scope) == 3 + assert Enum.map(Queue.pending(scope, 10), & &1.seq) == [3, 4, 5] + + assert TestAuthority.FailsFrom.seen() == [1, 2, 3], + "the forwarder offered rows after the one that failed" + end) + end + + test "the local authority acknowledges, so standalone Trinity exercises the same path", %{ + scope: scope + } do + for n <- 1..3, do: {:ok, _} = Receipts.append(scope, decision(n)) + assert Queue.depth(scope) == 3 + + assert %{acked: 3, failed: 0} = Forwarder.drain(scope) + assert Queue.depth(scope) == 0 + end + end + + describe "acknowledgement is in order, and the queue says so" do + test "acking a later receipt while an older one is pending is refused", %{scope: scope} do + for n <- 1..3, do: {:ok, _} = Receipts.append(scope, decision(n)) + [_first, second, _third] = Queue.pending(scope, 10) + + assert {:error, {:out_of_order, 1}} = Queue.ack(scope, second.receipt_hash) + assert Queue.depth(scope) == 3 + end + + test "acking in order works, and acking twice is idempotent", %{scope: scope} do + for n <- 1..2, do: {:ok, _} = Receipts.append(scope, decision(n)) + [first, second] = Queue.pending(scope, 10) + + assert :ok = Queue.ack(scope, first.receipt_hash) + assert :ok = Queue.ack(scope, first.receipt_hash) + assert :ok = Queue.ack(scope, second.receipt_hash) + assert Queue.depth(scope) == 0 + end + + test "acking something never queued is an error, not a silent success", %{scope: scope} do + assert {:error, :not_queued} = Queue.ack(scope, "no-such-hash") + end + end + + describe "AC4: past the bound, effects are denied rather than left unacknowledged" do + test "an append past the bound is refused", %{scope: scope} do + with_bound(3, fn -> + for n <- 1..3, do: {:ok, _} = Receipts.append(scope, decision(n)) + + assert {:error, {:queue_full, 3, 3}} = Receipts.append(scope, decision(4)) + end) + end + + test "the refusal is receipted, and that receipt is NOT itself queued", %{scope: scope} do + with_bound(2, fn -> + for n <- 1..2, do: {:ok, _} = Receipts.append(scope, decision(n)) + {:error, {:queue_full, _, _}} = Receipts.append(scope, decision(3)) + + refusal = + scope + |> Receipts.list() + |> Enum.find(&(&1.subject_ref && String.starts_with?(&1.subject_ref, "queue_full:"))) + + assert refusal, "the machine refused an effect and recorded nothing" + + {:ok, body} = JSON.decode(refusal.signed_payload) + assert get_in(body, ["decision", "outcome"]) == "refused" + assert get_in(body, ["decision", "basis"]) == "queue" + assert get_in(body, ["subject", "queue_bound"]) == 2 + + refute RRepo.get_by(QueueEntry, receipt_hash: refusal.receipt_hash), + "the queue-full refusal was itself queued. A full queue then cannot record that " <> + "it is full, which is a deadlock dressed as a safety property." + + assert Queue.depth(scope) == 2, "the refusal pushed the queue past its own bound" + end) + end + + test "the refused effect is not in the chain", %{scope: scope} do + with_bound(2, fn -> + for n <- 1..2, do: {:ok, _} = Receipts.append(scope, decision(n)) + {:error, _} = Receipts.append(scope, decision(3)) + + refute Enum.any?(Receipts.list(scope), fn r -> + {:ok, body} = JSON.decode(r.signed_payload) + body["fingerprint"] == "ab3" + end), + "an effect was refused and written anyway" + end) + end + + test "draining the queue lets effects proceed again", %{scope: scope} do + with_bound(2, fn -> + for n <- 1..2, do: {:ok, _} = Receipts.append(scope, decision(n)) + assert {:error, {:queue_full, _, _}} = Receipts.append(scope, decision(3)) + + assert %{acked: 2} = Forwarder.drain(scope) + assert {:ok, _} = Receipts.append(scope, decision(4)) + end) + end + + test "the chain still verifies across a refusal", %{scope: scope} do + with_bound(2, fn -> + for n <- 1..2, do: {:ok, _} = Receipts.append(scope, decision(n)) + {:error, _} = Receipts.append(scope, decision(3)) + end) + + :ok = Receipts.stop_writer(scope) + {:ok, export} = Receipts.export(scope) + assert {:ok, _} = Trinity.Receipts.Verifier.verify(export) + end + end +end diff --git a/test/trinity/receipts/scheme_mapping_test.exs b/test/trinity/receipts/scheme_mapping_test.exs index 7e5861d..0d6cefc 100644 --- a/test/trinity/receipts/scheme_mapping_test.exs +++ b/test/trinity/receipts/scheme_mapping_test.exs @@ -22,6 +22,11 @@ defmodule Trinity.Receipts.SchemeMappingTest do # derived from the code would agree with the code by construction and could prove nothing. @slice_024_keys ~w(scheme seq chain_scope prev_hash kind subject decision fingerprint at key_id) + # Slice 026 added exactly one field, under a new scheme version. Written out for the same reason, + # and as the whole of the difference: a future slice that adds a field silently fails here. + @slice_026_added ~w(clock) + @current_keys @slice_024_keys ++ @slice_026_added + defp payload_keys do scope = "test:" <> Trinity.UUID.generate() on_exit(fn -> Receipts.stop_writer(scope) end) @@ -51,8 +56,17 @@ defmodule Trinity.Receipts.SchemeMappingTest do |> Enum.sort() end - test "AC2: the signed payload's keys are exactly the ten slice 024 established" do - assert payload_keys() == Enum.sort(@slice_024_keys) + test "AC2: the signed payload's keys are slice 024's ten plus exactly what slice 026 added" do + assert payload_keys() == Enum.sort(@current_keys) + end + + test "AC2: slice 026 added one field and removed none, so no v2 byte could have changed" do + assert @slice_024_keys -- @current_keys == [], + "a field slice 024 signed is no longer in the payload. A scheme bump adds; it does " <> + "not drop, because a dropped field changes what an old row means when it is read " <> + "back by code that expects it." + + assert @current_keys -- @slice_024_keys == @slice_026_added end test "AC3: every signed field has a row in the mapping table, and every row names a signed field" do