21 KiB
21 KiB
Deneb -- Polynomial Commitments
Table of contents
- Introduction
- Custom types
- Constants
- Preset
- Helper functions
- Cells
- Reconstruction
Introduction
This document extends polynomial-commitments.md with the functions required for data availability sampling (DAS). It is not part of the core Deneb spec but an extension that can be optionally implemented to allow nodes to reduce their load using DAS.
For any KZG library extended to support DAS, functions flagged as "Public method" MUST be provided by the underlying KZG library as public functions. All other functions are private functions used internally by the KZG library.
Public functions MUST accept raw bytes as input and perform the required cryptographic normalization before invoking any internal functions.
Custom types
| Name | SSZ equivalent | Description |
|---|---|---|
PolynomialCoeff |
List[BLSFieldElement, 2 * FIELD_ELEMENTS_PER_BLOB] |
A polynomial in coefficient form |
Cell |
Vector[BLSFieldElement, FIELD_ELEMENTS_PER_CELL] |
The unit of blob data that can come with their own KZG proofs |
CellID |
uint64 |
Cell identifier |
Constants
| Name | Value | Notes |
|---|
Preset
Cells
Cells are the smallest unit of blob data that can come with their own KZG proofs. Samples can be constructed from one or several cells (e.g. an individual cell or line).
| Name | Value | Description |
|---|---|---|
FIELD_ELEMENTS_PER_CELL |
uint64(64) |
Number of field elements in a cell |
BYTES_PER_CELL |
FIELD_ELEMENTS_PER_CELL * BYTES_PER_FIELD_ELEMENT |
The number of bytes in a cell |
CELLS_PER_BLOB |
((2 * FIELD_ELEMENTS_PER_BLOB) // FIELD_ELEMENTS_PER_CELL) |
The number of cells in a blob |
RANDOM_CHALLENGE_KZG_CELL_BATCH_DOMAIN |
b'RCKZGCBATCH__V1_' |
Helper functions
BLS12-381 helpers
bytes_to_cell
def bytes_to_cell(cell_bytes: Vector[Bytes32, FIELD_ELEMENTS_PER_CELL]) -> Cell:
"""
Convert untrusted bytes into a Cell.
"""
return [bytes_to_bls_field(element) for element in cell_bytes]
Linear combinations
g2_lincomb
def g2_lincomb(points: Sequence[KZGCommitment], scalars: Sequence[BLSFieldElement]) -> Bytes96:
"""
BLS multiscalar multiplication in G2. This function can be optimized using Pippenger's algorithm and variants.
"""
assert len(points) == len(scalars)
result = bls.Z2()
for x, a in zip(points, scalars):
result = bls.add(result, bls.multiply(bls.bytes96_to_G2(x), a))
return Bytes96(bls.G2_to_bytes96(result))
FFTs
_fft_field
def _fft_field(vals: Sequence[BLSFieldElement],
roots_of_unity: Sequence[BLSFieldElement]) -> Sequence[BLSFieldElement]:
if len(vals) == 1:
return vals
L = _fft_field(vals[::2], roots_of_unity[::2])
R = _fft_field(vals[1::2], roots_of_unity[::2])
o = [BLSFieldElement(0) for _ in vals]
for i, (x, y) in enumerate(zip(L, R)):
y_times_root = (int(y) * int(roots_of_unity[i])) % BLS_MODULUS
o[i] = BLSFieldElement((int(x) + y_times_root) % BLS_MODULUS)
o[i + len(L)] = BLSFieldElement((int(x) - y_times_root + BLS_MODULUS) % BLS_MODULUS)
return o
fft_field
def fft_field(vals: Sequence[BLSFieldElement],
roots_of_unity: Sequence[BLSFieldElement],
inv: bool=False) -> Sequence[BLSFieldElement]:
if inv:
# Inverse FFT
invlen = pow(len(vals), BLS_MODULUS - 2, BLS_MODULUS)
return [BLSFieldElement((int(x) * invlen) % BLS_MODULUS)
for x in _fft_field(vals, list(roots_of_unity[0:1]) + list(roots_of_unity[:0:-1]))]
else:
# Regular FFT
return _fft_field(vals, roots_of_unity)
Polynomials in coefficient form
polynomial_eval_to_coeff
def polynomial_eval_to_coeff(polynomial: Polynomial) -> PolynomialCoeff:
"""
Interpolates a polynomial (given in evaluation form) to a polynomial in coefficient form.
"""
roots_of_unity = compute_roots_of_unity(FIELD_ELEMENTS_PER_BLOB)
polynomial_coeff = fft_field(bit_reversal_permutation(list(polynomial)), roots_of_unity, inv=True)
return polynomial_coeff
add_polynomialcoeff
def add_polynomialcoeff(a: PolynomialCoeff, b: PolynomialCoeff) -> PolynomialCoeff:
"""
Sum the coefficient form polynomials ``a`` and ``b``.
"""
a, b = (a, b) if len(a) >= len(b) else (b, a)
length_a = len(a)
length_b = len(b)
return [(a[i] + (b[i] if i < length_b else 0)) % BLS_MODULUS for i in range(length_a)]
neg_polynomialcoeff
def neg_polynomialcoeff(a: PolynomialCoeff) -> PolynomialCoeff:
"""
Negative of coefficient form polynomial ``a``
"""
return [(BLS_MODULUS - x) % BLS_MODULUS for x in a]
multiply_polynomialcoeff
def multiply_polynomialcoeff(a: PolynomialCoeff, b: PolynomialCoeff) -> PolynomialCoeff:
"""
Multiplies the coefficient form polynomials ``a`` and ``b``
"""
r = [0]
for power, coef in enumerate(a):
summand = [0] * power + [int(coef) * int(x) % BLS_MODULUS for x in b]
r = add_polynomialcoeff(r, summand)
return r
divide_polynomialcoeff
def divide_polynomialcoeff(a: PolynomialCoeff, b: PolynomialCoeff) -> PolynomialCoeff:
"""
Long polynomial division for two coefficient form polynomials ``a`` and ``b``
"""
a = [x for x in a]
o = []
apos = len(a) - 1
bpos = len(b) - 1
diff = apos - bpos
while diff >= 0:
quot = div(a[apos], b[bpos])
o.insert(0, quot)
for i in range(bpos, -1, -1):
a[diff + i] = (int(a[diff + i]) - int(b[i]) * int(quot)) % BLS_MODULUS
apos -= 1
diff -= 1
return [x % BLS_MODULUS for x in o]
shift_polynomialcoeff
def shift_polynomialcoeff(polynomial_coeff: PolynomialCoeff, factor: BLSFieldElement) -> PolynomialCoeff:
"""
Shift the evaluation of a polynomial in coefficient form by factor.
This results in a new polynomial g(x) = f(factor * x)
"""
factor_power = 1
inv_factor = pow(int(factor), BLS_MODULUS - 2, BLS_MODULUS)
o = []
for p in polynomial_coeff:
o.append(int(p) * factor_power % BLS_MODULUS)
factor_power = factor_power * inv_factor % BLS_MODULUS
return o
interpolate_polynomialcoeff
def interpolate_polynomialcoeff(xs: Sequence[BLSFieldElement], ys: Sequence[BLSFieldElement]) -> PolynomialCoeff:
"""
Lagrange interpolation: Finds the lowest degree polynomial that takes the value ``ys[i]`` at ``x[i]``
for all i.
Outputs a coefficient form polynomial. Leading coefficients may be zero.
"""
assert len(xs) == len(ys)
r = [0]
for i in range(len(xs)):
summand = [ys[i]]
for j in range(len(ys)):
if j != i:
weight_adjustment = bls_modular_inverse(int(xs[i]) - int(xs[j]))
summand = multiply_polynomialcoeff(
summand, [(- int(weight_adjustment) * int(xs[j])) % BLS_MODULUS, weight_adjustment]
)
r = add_polynomialcoeff(r, summand)
return r
vanishing_polynomialcoeff
def vanishing_polynomialcoeff(xs: Sequence[BLSFieldElement]) -> PolynomialCoeff:
"""
Compute the vanishing polynomial on ``xs`` (in coefficient form)
"""
p = [1]
for x in xs:
p = multiply_polynomialcoeff(p, [-int(x), 1])
return p
evaluate_polynomialcoeff
def evaluate_polynomialcoeff(polynomial_coeff: PolynomialCoeff, z: BLSFieldElement) -> BLSFieldElement:
"""
Evaluate a coefficient form polynomial at ``z`` using Horner's schema
"""
y = 0
for coef in polynomial_coeff[::-1]:
y = (int(y) * int(z) + int(coef)) % BLS_MODULUS
return BLSFieldElement(y % BLS_MODULUS)
KZG multiproofs
Extended KZG functions for multiproofs
compute_kzg_proof_multi_impl
def compute_kzg_proof_multi_impl(
polynomial_coeff: PolynomialCoeff,
zs: Sequence[BLSFieldElement]) -> Tuple[KZGProof, Sequence[BLSFieldElement]]:
"""
Helper function that computes multi-evaluation KZG proofs.
"""
# For all x_i, compute p(x_i) - p(z)
ys = [evaluate_polynomialcoeff(polynomial_coeff, z) for z in zs]
interpolation_polynomial = interpolate_polynomialcoeff(zs, ys)
polynomial_shifted = add_polynomialcoeff(polynomial_coeff, neg_polynomialcoeff(interpolation_polynomial))
# For all x_i, compute (x_i - z)
denominator_poly = vanishing_polynomialcoeff(zs)
# Compute the quotient polynomial directly in evaluation form
quotient_polynomial = divide_polynomialcoeff(polynomial_shifted, denominator_poly)
return KZGProof(g1_lincomb(KZG_SETUP_G1_MONOMIAL[:len(quotient_polynomial)], quotient_polynomial)), ys
verify_kzg_proof_multi_impl
def verify_kzg_proof_multi_impl(commitment: KZGCommitment,
zs: Sequence[BLSFieldElement],
ys: Sequence[BLSFieldElement],
proof: KZGProof) -> bool:
"""
Helper function that verifies a KZG multiproof
"""
assert len(zs) == len(ys)
zero_poly = g2_lincomb(KZG_SETUP_G2_MONOMIAL[:len(zs) + 1], vanishing_polynomialcoeff(zs))
interpolated_poly = g1_lincomb(KZG_SETUP_G1_MONOMIAL[:len(zs)], interpolate_polynomialcoeff(zs, ys))
return (bls.pairing_check([
[bls.bytes48_to_G1(proof), bls.bytes96_to_G2(zero_poly)],
[
bls.add(bls.bytes48_to_G1(commitment), bls.neg(bls.bytes48_to_G1(interpolated_poly))),
bls.neg(bls.bytes96_to_G2(KZG_SETUP_G2_MONOMIAL[0])),
],
]))
Cell cosets
coset_for_cell
def coset_for_cell(cell_id: CellID) -> Cell:
"""
Get the coset for a given ``cell_id``
"""
assert cell_id < CELLS_PER_BLOB
roots_of_unity_brp = bit_reversal_permutation(
compute_roots_of_unity(2 * FIELD_ELEMENTS_PER_BLOB)
)
return Cell(roots_of_unity_brp[FIELD_ELEMENTS_PER_CELL * cell_id:FIELD_ELEMENTS_PER_CELL * (cell_id + 1)])
Cells
Cell computation
compute_cells_and_proofs
def compute_cells_and_proofs(blob: Blob) -> Tuple[
Vector[Cell, CELLS_PER_BLOB],
Vector[KZGProof, CELLS_PER_BLOB]]:
"""
Compute all the cell proofs for one blob. This is an inefficient O(n^2) algorithm,
for performant implementation the FK20 algorithm that runs in O(n log n) should be
used instead.
Public method.
"""
polynomial = blob_to_polynomial(blob)
polynomial_coeff = polynomial_eval_to_coeff(polynomial)
cells = []
proofs = []
for i in range(CELLS_PER_BLOB):
coset = coset_for_cell(i)
proof, ys = compute_kzg_proof_multi_impl(polynomial_coeff, coset)
cells.append(ys)
proofs.append(proof)
return cells, proofs
compute_cells
def compute_cells(blob: Blob) -> Vector[Cell, CELLS_PER_BLOB]:
"""
Compute the cell data for a blob (without computing the proofs).
Public method.
"""
polynomial = blob_to_polynomial(blob)
polynomial_coeff = polynomial_eval_to_coeff(polynomial)
extended_data = fft_field(polynomial_coeff + [0] * FIELD_ELEMENTS_PER_BLOB,
compute_roots_of_unity(2 * FIELD_ELEMENTS_PER_BLOB))
extended_data_rbo = bit_reversal_permutation(extended_data)
return [extended_data_rbo[i * FIELD_ELEMENTS_PER_CELL:(i + 1) * FIELD_ELEMENTS_PER_CELL]
for i in range(CELLS_PER_BLOB)]
Cell verification
verify_cell_proof
def verify_cell_proof(commitment_bytes: Bytes48,
cell_id: CellID,
cell_bytes: Vector[Bytes32, FIELD_ELEMENTS_PER_CELL],
proof_bytes: Bytes48) -> bool:
"""
Check a cell proof
Public method.
"""
coset = coset_for_cell(cell_id)
return verify_kzg_proof_multi_impl(
bytes_to_kzg_commitment(commitment_bytes),
coset,
bytes_to_cell(cell_bytes),
bytes_to_kzg_proof(proof_bytes))
verify_cell_proof_batch
def verify_cell_proof_batch(row_commitments_bytes: Sequence[Bytes48],
row_ids: Sequence[uint64],
column_ids: Sequence[uint64],
cells_bytes: Sequence[Vector[Bytes32, FIELD_ELEMENTS_PER_CELL]],
proofs_bytes: Sequence[Bytes48]) -> bool:
"""
Verify a set of cells, given their corresponding proofs and their coordinates (row_id, column_id) in the blob
matrix. The list of all commitments is also provided in row_commitments_bytes.
This function implements the naive algorithm of checking every cell
individually; an efficient algorithm can be found here:
https://ethresear.ch/t/a-universal-verification-equation-for-data-availability-sampling/13240
This implementation does not require randomness, but for the algorithm that
requires it, `RANDOM_CHALLENGE_KZG_CELL_BATCH_DOMAIN` should be used to compute
the challenge value.
Public method.
"""
assert len(cells_bytes) == len(proofs_bytes) == len(row_ids) == len(column_ids)
# Get commitments via row IDs
commitments_bytes = [row_commitments_bytes[row_id] for row_id in row_ids]
# Get objects from bytes
commitments = [bytes_to_kzg_commitment(commitment_bytes) for commitment_bytes in commitments_bytes]
cells = [bytes_to_cell(cell_bytes) for cell_bytes in cells_bytes]
proofs = [bytes_to_kzg_proof(proof_bytes) for proof_bytes in proofs_bytes]
return all(
verify_kzg_proof_multi_impl(commitment, coset_for_cell(column_id), cell, proof)
for commitment, column_id, cell, proof in zip(commitments, column_ids, cells, proofs)
)
Reconstruction
construct_vanishing_polynomial
def construct_vanishing_polynomial(cell_ids: Sequence[CellID],
cells: Sequence[Cell]) -> Tuple[
Sequence[BLSFieldElement],
Sequence[BLSFieldElement]]:
missing_cell_ids = [cell_id for cell_id in range(CELLS_PER_BLOB) if cell_id not in cell_ids]
roots_of_unity_reduced = compute_roots_of_unity(CELLS_PER_BLOB)
short_zero_poly = vanishing_polynomialcoeff([
roots_of_unity_reduced[reverse_bits(cell_id, CELLS_PER_BLOB)]
for cell_id in missing_cell_ids
])
zero_poly_coeff = []
for i in short_zero_poly:
zero_poly_coeff.append(i)
zero_poly_coeff.extend([0] * (FIELD_ELEMENTS_PER_CELL - 1))
zero_poly_coeff = zero_poly_coeff + [0] * (2 * FIELD_ELEMENTS_PER_BLOB - len(zero_poly_coeff))
zero_poly_eval = fft_field(zero_poly_coeff,
compute_roots_of_unity(2 * FIELD_ELEMENTS_PER_BLOB))
zero_poly_eval_brp = bit_reversal_permutation(zero_poly_eval)
for cell_id in missing_cell_ids:
start = cell_id * FIELD_ELEMENTS_PER_CELL
end = (cell_id + 1) * FIELD_ELEMENTS_PER_CELL
assert zero_poly_eval_brp[start:end] == [0] * FIELD_ELEMENTS_PER_CELL
for cell_id in cell_ids:
start = cell_id * FIELD_ELEMENTS_PER_CELL
end = (cell_id + 1) * FIELD_ELEMENTS_PER_CELL
assert all(a != 0 for a in zero_poly_eval_brp[start:end])
return zero_poly_coeff, zero_poly_eval, zero_poly_eval_brp
recover_shifted_data
def recover_shifted_data(cell_ids: Sequence[CellID],
cells: Sequence[Cell],
zero_poly_eval: Sequence[BLSFieldElement],
zero_poly_coeff: Sequence[BLSFieldElement],
roots_of_unity_extended: Sequence[BLSFieldElement]) -> Tuple[
Sequence[BLSFieldElement],
Sequence[BLSFieldElement],
BLSFieldElement]:
extended_evaluation_rbo = [0] * (FIELD_ELEMENTS_PER_BLOB * 2)
for cell_id, cell in zip(cell_ids, cells):
start = cell_id * FIELD_ELEMENTS_PER_CELL
end = (cell_id + 1) * FIELD_ELEMENTS_PER_CELL
extended_evaluation_rbo[start:end] = cell
extended_evaluation = bit_reversal_permutation(extended_evaluation_rbo)
extended_evaluation_times_zero = [BLSFieldElement(int(a) * int(b) % BLS_MODULUS)
for a, b in zip(zero_poly_eval, extended_evaluation)]
extended_evaluations_fft = fft_field(extended_evaluation_times_zero, roots_of_unity_extended, inv=True)
shift_factor = BLSFieldElement(PRIMITIVE_ROOT_OF_UNITY)
shift_inv = div(BLSFieldElement(1), shift_factor)
shifted_extended_evaluation = shift_polynomialcoeff(extended_evaluations_fft, shift_factor)
shifted_zero_poly = shift_polynomialcoeff(zero_poly_coeff, shift_factor)
eval_shifted_extended_evaluation = fft_field(shifted_extended_evaluation, roots_of_unity_extended)
eval_shifted_zero_poly = fft_field(shifted_zero_poly, roots_of_unity_extended)
return eval_shifted_extended_evaluation, eval_shifted_zero_poly, shift_inv
recover_original_data
def recover_original_data(eval_shifted_extended_evaluation: Sequence[BLSFieldElement],
eval_shifted_zero_poly: Sequence[BLSFieldElement],
shift_inv: BLSFieldElement,
roots_of_unity_extended: Sequence[BLSFieldElement]) -> Sequence[BLSFieldElement]:
eval_shifted_reconstructed_poly = [
div(a, b)
for a, b in zip(eval_shifted_extended_evaluation, eval_shifted_zero_poly)
]
shifted_reconstructed_poly = fft_field(eval_shifted_reconstructed_poly, roots_of_unity_extended, inv=True)
reconstructed_poly = shift_polynomialcoeff(shifted_reconstructed_poly, shift_inv)
reconstructed_data = bit_reversal_permutation(fft_field(reconstructed_poly, roots_of_unity_extended))
return reconstructed_data
recover_polynomial
def recover_polynomial(cell_ids: Sequence[CellID],
cells_bytes: Sequence[Vector[Bytes32, FIELD_ELEMENTS_PER_CELL]]) -> Polynomial:
"""
Recovers a polynomial from 2 * FIELD_ELEMENTS_PER_CELL evaluations, half of which can be missing.
This algorithm uses FFTs to recover cells faster than using Lagrange implementation. However,
a faster version thanks to Qi Zhou can be found here:
https://github.com/ethereum/research/blob/51b530a53bd4147d123ab3e390a9d08605c2cdb8/polynomial_reconstruction/polynomial_reconstruction_danksharding.py
Public method.
"""
assert len(cell_ids) == len(cells_bytes)
cells = [bytes_to_cell(cell_bytes) for cell_bytes in cells_bytes]
assert len(cells) >= CELLS_PER_BLOB // 2
roots_of_unity_extended = compute_roots_of_unity(2 * FIELD_ELEMENTS_PER_BLOB)
zero_poly_coeff, zero_poly_eval, zero_poly_eval_brp = construct_vanishing_polynomial(cell_ids, cells)
eval_shifted_extended_evaluation, eval_shifted_zero_poly, shift_inv = \
recover_shifted_data(cell_ids, cells, zero_poly_eval, zero_poly_coeff, roots_of_unity_extended)
reconstructed_data = \
recover_original_data(eval_shifted_extended_evaluation, eval_shifted_zero_poly, shift_inv,
roots_of_unity_extended)
for cell_id, cell in zip(cell_ids, cells):
start = cell_id * FIELD_ELEMENTS_PER_CELL
end = (cell_id + 1) * FIELD_ELEMENTS_PER_CELL
assert reconstructed_data[start:end] == cell
return reconstructed_data