cumulus_primitives_spec_messaging/
streams_root.rs

1// Copyright (C) Parity Technologies (UK) Ltd.
2// This file is part of Cumulus.
3// SPDX-License-Identifier: Apache-2.0
4
5// Licensed under the Apache License, Version 2.0 (the "License");
6// you may not use this file except in compliance with the License.
7// You may obtain a copy of the License at
8//
9// 	http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing, software
12// distributed under the License is distributed on an "AS IS" BASIS,
13// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14// See the License for the specific language governing permissions and
15// limitations under the License.
16
17//! `StreamsRoot`: the sender's stream commitment.
18//!
19//! Each block that sends commits one hash, the root of a keyed binary Patricia trie over
20//! `(StreamId, current stream root)` for every active stream. The relay records it as the sender's
21//! `Provides`, and a receiver proves the streams it consumed against it, without a sender-state
22//! proof.
23//!
24//! # Why a keyed trie
25//!
26//! The trie is keyed by [`StreamId::to_bytes`], so a proof is a walk driven by the key:
27//! verification recomputes the leaf from the caller's `StreamId` and takes each branch direction
28//! from that key's bits. A proof built for stream A cannot verify as stream B. Requires-lift
29//! synthesis depends on this, since lifts are matched to the consumption record's streams by
30//! position. A positional Merkle tree does not give it. Patricia compression also keeps proofs
31//! `O(log S)` in the number of streams rather than `O(64)` in the key width.
32//!
33//! Roles: the sender builds the root ([`streams_root`]) and its node serves proofs
34//! ([`gen_stream_proof`]), checking its own header digest ([`read_streams_root`]); a receiver's
35//! lift walks a proof up to the committed root ([`streams_root_from_proof`]).
36//!
37//! A leaf hash binds the full key, an inner hash its split bit.
38
39use alloc::{collections::BTreeMap, vec::Vec};
40use codec::{Decode, DecodeAll, DecodeWithMemTracking, Encode};
41use polkadot_core_primitives::Hash;
42use sp_core::ConstU32;
43use sp_runtime::{traits::Hash as HashT, BoundedVec, Digest, DigestItem};
44
45use crate::{
46	stream::StreamId, SpecHasher, SPMS_ENGINE_ID, STREAM_ID_LEN, TREE_INNER_TAG, TREE_LEAF_TAG,
47};
48
49/// Bit width of a `StreamId` key (the Patricia trie's maximum depth).
50const KEY_BITS: usize = STREAM_ID_LEN * 8;
51
52/// The sender's per-block stream commitment root. Defined relay-side and re-exported here.
53pub use polkadot_primitives::StreamsRoot;
54
55/// One step of a trie walk. The branch direction is not stored; verification takes it from the
56/// proven key's bit at `split_bit`.
57#[derive(
58	Clone, Copy, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, scale_info::TypeInfo,
59)]
60pub struct TreeStep {
61	/// Key-bit index this node branches on, `0..KEY_BITS` with `0` the MSB. Part of the inner-node
62	/// hash; the proven key's bit here gives the direction.
63	pub split_bit: u8,
64	pub sibling: Hash,
65}
66
67/// A membership proof that `(stream, stream_root)` is committed under a `StreamsRoot`. Served by
68/// the sender, verified by the receiver. Carries no leaf index; the key drives the walk.
69#[derive(
70	Clone, Encode, Decode, DecodeWithMemTracking, PartialEq, Eq, Debug, scale_info::TypeInfo,
71)]
72pub struct StreamProof {
73	/// Branch steps leaf to root, split bits strictly decreasing. Empty for a single-stream tree.
74	/// Bounded at the trie depth, so an over-long proof is rejected at decode.
75	pub steps: BoundedVec<TreeStep, ConstU32<{ KEY_BITS as u32 }>>,
76}
77
78/// Bit `i` of an 8-byte key, `0` the MSB of byte 0.
79fn bit(key: &[u8; STREAM_ID_LEN], i: usize) -> bool {
80	(key[i >> 3] >> (7 - (i & 7))) & 1 == 1
81}
82
83/// First bit index (from the MSB) at which two distinct keys differ. Callers guarantee `a != b`.
84fn first_diverging_bit(a: &[u8; STREAM_ID_LEN], b: &[u8; STREAM_ID_LEN]) -> usize {
85	(u64::from_be_bytes(*a) ^ u64::from_be_bytes(*b)).leading_zeros() as usize
86}
87
88fn hash_leaf(key: &[u8; STREAM_ID_LEN], stream_root: &Hash) -> Hash {
89	let mut pre = [0u8; 1 + STREAM_ID_LEN + 32];
90	pre[0] = TREE_LEAF_TAG;
91	pre[1..1 + STREAM_ID_LEN].copy_from_slice(key);
92	pre[1 + STREAM_ID_LEN..].copy_from_slice(stream_root.as_bytes());
93	<SpecHasher as HashT>::hash(&pre)
94}
95
96fn hash_inner(split_bit: u8, left: &Hash, right: &Hash) -> Hash {
97	let mut pre = [0u8; 1 + 1 + 32 + 32];
98	pre[0] = TREE_INNER_TAG;
99	pre[1] = split_bit;
100	pre[2..34].copy_from_slice(left.as_bytes());
101	pre[34..].copy_from_slice(right.as_bytes());
102	<SpecHasher as HashT>::hash(&pre)
103}
104
105/// Split a key-sorted slice at its top-level branch: the bit where its first and last keys diverge,
106/// with all `bit == 0` keys before all `bit == 1` keys. Both sides are non-empty. Callers guarantee
107/// `entries.len() >= 2` and distinct keys.
108fn split<'a>(
109	entries: &'a [([u8; STREAM_ID_LEN], Hash)],
110) -> (usize, &'a [([u8; STREAM_ID_LEN], Hash)], &'a [([u8; STREAM_ID_LEN], Hash)]) {
111	let split_bit = first_diverging_bit(&entries[0].0, &entries[entries.len() - 1].0);
112	let p = entries.partition_point(|(k, _)| !bit(k, split_bit));
113	let (left, right) = entries.split_at(p);
114	(split_bit, left, right)
115}
116
117/// Trie root over a key-sorted, non-empty slice.
118fn node_hash(entries: &[([u8; STREAM_ID_LEN], Hash)]) -> Hash {
119	if let [(key, root)] = entries {
120		return hash_leaf(key, root);
121	}
122	let (split_bit, left, right) = split(entries);
123	hash_inner(split_bit as u8, &node_hash(left), &node_hash(right))
124}
125
126/// Encoded keys in trie order.
127///
128/// The trie splits on bytes, so sort by them rather than rely on `Ord`. Unsorted input would leave
129/// one side of `split` empty and make `prove` recurse on its own slice.
130fn keyed(entries: &BTreeMap<StreamId, Hash>) -> Vec<([u8; STREAM_ID_LEN], Hash)> {
131	let mut keyed: Vec<_> =
132		entries.iter().map(|(stream, root)| (stream.to_bytes(), *root)).collect();
133	keyed.sort_unstable_by_key(|(key, _)| *key);
134	keyed
135}
136
137/// The `StreamsRoot` over `(stream, stream_root)` entries; `None` with no active streams. Sender
138/// side. Takes a `BTreeMap` so keys are unique; duplicates would reach `first_diverging_bit` with
139/// identical keys.
140///
141/// TODO(spec-msg): update one changed stream in O(log S) instead of rebuilding in O(S); the sender
142/// pallet rebuilds this every touched block over all its streams.
143pub fn streams_root(entries: &BTreeMap<StreamId, Hash>) -> Option<StreamsRoot> {
144	let entries = keyed(entries);
145	if entries.is_empty() {
146		return None;
147	}
148	Some(StreamsRoot(node_hash(&entries)))
149}
150
151/// Walk key-sorted `entries` toward `key`, pushing each branch step root-first, and return the
152/// subtree's root. `key` must be present.
153fn prove(
154	entries: &[([u8; STREAM_ID_LEN], Hash)],
155	key: &[u8; STREAM_ID_LEN],
156	steps: &mut Vec<TreeStep>,
157) -> Hash {
158	if let [(k, root)] = entries {
159		return hash_leaf(k, root);
160	}
161	let (split_bit, left, right) = split(entries);
162	let target_right = bit(key, split_bit);
163	let (sibling, descend) =
164		if target_right { (node_hash(left), right) } else { (node_hash(right), left) };
165	steps.push(TreeStep { split_bit: split_bit as u8, sibling });
166	let child = prove(descend, key, steps);
167	if target_right {
168		hash_inner(split_bit as u8, &sibling, &child)
169	} else {
170		hash_inner(split_bit as u8, &child, &sibling)
171	}
172}
173
174/// Build the trie over `entries` and return `(StreamsRoot, proof)` for `stream`, or `None` if
175/// absent. Sender side.
176pub fn gen_stream_proof(
177	entries: &BTreeMap<StreamId, Hash>,
178	stream: StreamId,
179) -> Option<(StreamsRoot, StreamProof)> {
180	if !entries.contains_key(&stream) {
181		return None;
182	}
183	let entries = keyed(entries);
184	let key = stream.to_bytes();
185	let mut steps = Vec::new();
186	let root = prove(&entries, &key, &mut steps);
187	// `prove` pushes root-first; the wire order is leaf-to-root.
188	steps.reverse();
189	let steps = BoundedVec::try_from(steps).expect("trie depth <= KEY_BITS; qed");
190	Some((StreamsRoot(root), StreamProof { steps }))
191}
192
193/// Read the sender's `StreamsRoot` from its header digest. `None` if absent or not exactly what the
194/// sender deposits: one `SPMS` item with a 32-byte payload. Foreign readers must agree on
195/// validity, so nothing else is accepted.
196pub fn read_streams_root(digest: &Digest) -> Option<StreamsRoot> {
197	let mut items = digest.logs().iter().filter_map(|item| match item {
198		DigestItem::Consensus(id, payload) if *id == SPMS_ENGINE_ID => Some(payload),
199		_ => None,
200	});
201	let payload = items.next()?;
202	if items.next().is_some() {
203		return None;
204	}
205	StreamsRoot::decode_all(&mut &payload[..]).ok()
206}
207
208/// Fold a membership proof for `(stream, stream_root)` up to the `StreamsRoot` it implies, or
209/// `None` on a malformed step. Callers compare the root themselves (the requires-lift records it,
210/// the fetch checks it against `under`). Each branch direction is `stream`'s bit at `split_bit`, so
211/// a proof built for another stream folds the siblings on the wrong sides and yields a different
212/// root.
213pub fn streams_root_from_proof(
214	stream: StreamId,
215	stream_root: Hash,
216	membership: &StreamProof,
217) -> Option<StreamsRoot> {
218	let key = stream.to_bytes();
219	let mut node = hash_leaf(&key, &stream_root);
220	// Steps are leaf-to-root with strictly decreasing split bits; anything else is rejected.
221	// Uniqueness rests on the key-in-leaf and bit-in-inner commitments, not on this rule. The
222	// length bound is in the type.
223	let mut prev: Option<u8> = None;
224	for step in membership.steps.iter() {
225		let split_bit = step.split_bit as usize;
226		// An out-of-range branch would index the key out of bounds.
227		if split_bit >= KEY_BITS {
228			return None;
229		}
230		if prev.is_some_and(|p| step.split_bit >= p) {
231			return None;
232		}
233		prev = Some(step.split_bit);
234		// The branch direction is the proven key's bit at `split_bit`. This is the key binding.
235		node = if bit(&key, split_bit) {
236			hash_inner(step.split_bit, &step.sibling, &node)
237		} else {
238			hash_inner(step.split_bit, &node, &step.sibling)
239		};
240	}
241	Some(StreamsRoot(node))
242}
243
244#[cfg(test)]
245mod tests {
246	use super::*;
247	use crate::test_utils::ch;
248	use sp_core::H256;
249
250	fn root(b: u8) -> Hash {
251		H256::repeat_byte(b)
252	}
253
254	/// Whether `stream_root` is committed for `stream` under `streams_root`.
255	fn verify_stream_membership(
256		streams_root: StreamsRoot,
257		stream: StreamId,
258		stream_root: Hash,
259		membership: &StreamProof,
260	) -> bool {
261		streams_root_from_proof(stream, stream_root, membership) == Some(streams_root)
262	}
263	fn entries() -> BTreeMap<StreamId, Hash> {
264		// inserted out of order; the map canonicalizes
265		BTreeMap::from([
266			(ch(4000), root(0xB)),
267			(ch(2000), root(0xA)),
268			(ch(3000), root(0xC)),
269			(StreamId::Ack { recipient: 2000.into(), domain: 0, num: 0 }, root(0xD)),
270			(StreamId::Broadcast { domain: 1, subdomain: 0, num: 7 }, root(0xE)),
271		])
272	}
273
274	#[test]
275	fn streams_root_frozen_vector() {
276		// Frozen vector. Any change to the trie encoding (tags, `split_bit`, ordering, key binding)
277		// is a consensus break.
278		assert_eq!(
279			streams_root(&entries()).unwrap(),
280			StreamsRoot(H256(hex_literal::hex!(
281				"12de235fb6d96933d6ba770efe16f891c87fb8a88a75e37a384b88ba11dee0dc"
282			))),
283		);
284	}
285
286	#[test]
287	fn round_trip_membership_verifies() {
288		for (stream, sr) in entries() {
289			let (r, proof) = gen_stream_proof(&entries(), stream).unwrap();
290			assert_eq!(Some(r), streams_root(&entries()));
291			assert!(verify_stream_membership(r, stream, sr, &proof), "{stream:?} must verify");
292		}
293	}
294
295	#[test]
296	fn wrong_root_or_stream_or_value_fails() {
297		let (r, proof) = gen_stream_proof(&entries(), ch(3000)).unwrap();
298		// wrong stream_root for the (correct) stream
299		assert!(!verify_stream_membership(r, ch(3000), root(0xFF), &proof));
300		// right value/proof but claimed for a different stream (key binding must reject)
301		assert!(!verify_stream_membership(r, ch(2000), root(0xC), &proof));
302		// wrong streams root
303		assert!(!verify_stream_membership(StreamsRoot(root(0x00)), ch(3000), root(0xC), &proof));
304	}
305
306	#[test]
307	fn proof_is_not_transferable_between_streams() {
308		// A proof for one stream must not verify for another, even with that stream's own value.
309		let (r, proof) = gen_stream_proof(&entries(), ch(4000)).unwrap();
310		for (other, other_root) in entries() {
311			if other == ch(4000) {
312				continue;
313			}
314			assert!(
315				!verify_stream_membership(r, other, other_root, &proof),
316				"proof for ch(4000) must not verify as {other:?}"
317			);
318		}
319	}
320
321	#[test]
322	fn order_independent_root() {
323		// Insertion order cannot affect the root.
324		let reversed: BTreeMap<_, _> = entries().into_iter().rev().collect();
325		assert_eq!(streams_root(&entries()), streams_root(&reversed));
326	}
327
328	#[test]
329	fn single_stream_tree_has_empty_proof() {
330		let one = BTreeMap::from([(ch(2000), root(0xA))]);
331		let (r, proof) = gen_stream_proof(&one, ch(2000)).unwrap();
332		assert!(proof.steps.is_empty());
333		assert_eq!(Some(r), streams_root(&one));
334		assert!(verify_stream_membership(r, ch(2000), root(0xA), &proof));
335	}
336
337	#[test]
338	fn proof_is_logarithmic() {
339		// 64 clustered channels: proof length is O(log S), not the 64-bit key width.
340		let many: BTreeMap<_, _> = (0..64u32).map(|i| (ch(2000 + i), root(i as u8))).collect();
341		let (_r, proof) = gen_stream_proof(&many, ch(2000 + 33)).unwrap();
342		assert!(proof.steps.len() <= 7, "log2(64) = 6-ish, got {}", proof.steps.len());
343	}
344
345	#[test]
346	fn absent_stream_has_no_proof() {
347		assert!(gen_stream_proof(&entries(), ch(9999)).is_none());
348	}
349
350	#[test]
351	fn out_of_range_split_bit_is_rejected() {
352		let (r, mut proof) = gen_stream_proof(&entries(), ch(3000)).unwrap();
353		// Splice the bogus step in first (largest bit) so the range check, not the order check,
354		// rejects it.
355		let mut steps = vec![TreeStep { split_bit: 64, sibling: root(0) }];
356		steps.extend(proof.steps.iter().copied());
357		proof.steps = steps.try_into().unwrap();
358		assert!(!verify_stream_membership(r, ch(3000), root(0xC), &proof));
359	}
360
361	#[test]
362	fn over_long_proof_is_rejected_at_decode() {
363		// A valid proof never exceeds the trie depth, so longer input fails at decode.
364		let raw = vec![TreeStep { split_bit: 0, sibling: root(0) }; KEY_BITS + 1].encode();
365		assert!(StreamProof::decode(&mut &raw[..]).is_err());
366	}
367
368	#[test]
369	fn non_decreasing_step_order_is_rejected() {
370		// A reversed sequence is rejected even though every hash in it is genuine.
371		let (r, proof) = gen_stream_proof(&entries(), ch(3000)).unwrap();
372		assert!(proof.steps.len() >= 2, "need a multi-step proof");
373		let reversed: Vec<TreeStep> = proof.steps.iter().rev().copied().collect();
374		let bad = StreamProof { steps: reversed.try_into().unwrap() };
375		assert!(!verify_stream_membership(r, ch(3000), root(0xC), &bad));
376	}
377
378	#[test]
379	fn read_streams_root_extracts_digest() {
380		let r = streams_root(&entries()).unwrap();
381		let digest = Digest {
382			logs: vec![
383				DigestItem::Other(vec![1, 2, 3]),
384				DigestItem::Consensus(SPMS_ENGINE_ID, r.encode()),
385			],
386		};
387		assert_eq!(read_streams_root(&digest), Some(r));
388		assert_eq!(read_streams_root(&Digest { logs: vec![] }), None);
389	}
390
391	#[test]
392	fn read_streams_root_accepts_only_the_canonical_digest() {
393		let r = streams_root(&entries()).unwrap();
394		let spms = |payload: Vec<u8>| DigestItem::Consensus(SPMS_ENGINE_ID, payload);
395		let digest = |logs: Vec<DigestItem>| Digest { logs };
396
397		// Exactly 32 bytes.
398		let mut trailing = r.encode();
399		trailing.push(0);
400		assert_eq!(read_streams_root(&digest(vec![spms(trailing)])), None);
401		assert_eq!(read_streams_root(&digest(vec![spms(r.encode()[..31].to_vec())])), None);
402
403		// Exactly one item. A malformed first item is not skipped for a later one.
404		let other = StreamsRoot(Hash::repeat_byte(0xAB));
405		assert_eq!(read_streams_root(&digest(vec![spms(r.encode()), spms(other.encode())])), None);
406		assert_eq!(read_streams_root(&digest(vec![spms(vec![0; 3]), spms(r.encode())])), None);
407
408		// Other engines' items are not counted against it.
409		let foreign = DigestItem::Consensus(*b"AURA", vec![0; 32]);
410		assert_eq!(read_streams_root(&digest(vec![foreign, spms(r.encode())])), Some(r));
411	}
412}
cumulus_primitives_spec_messaging/streams_root.rs

cumulus_primitives_spec_messaging/
streams_root.rs
