Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 15 additions & 43 deletions src/codex32/bip93.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

from collections.abc import Sequence
from dataclasses import dataclass
from itertools import islice

from codex32.bech32 import (
CHARSET,
Expand Down Expand Up @@ -133,19 +134,15 @@ def _decode_codex32(text: str) -> tuple[str, tuple[int, ...], _Checksum]:

@dataclass(frozen=True, slots=True, init=False)
class _Artifact:
"""Create artifacts with parse_codex32 or the factories, which validate them."""

text: str
header: Header
hrp: str
payload_symbols: tuple[int, ...]

def __init__(
self,
text: str,
header: Header,
hrp: str,
payload_symbols: tuple[int, ...],
*,
_token: object,
self, text: str, header: Header, hrp: str, payload_symbols: tuple[int, ...], *, _token: object
) -> None:
if _token is not _CONSTRUCTION_TOKEN:
raise TypeError("codex32 artifacts must be created by the public factories")
Expand Down Expand Up @@ -262,43 +259,24 @@ def indices(self) -> tuple[str, ...]:
return tuple(item.header.index for item in self.artifacts)


def _bounded_artifacts(
artifacts: Sequence[Share | Secret],
) -> tuple[Share | Secret, ...]:
if isinstance(artifacts, (str, bytes, bytearray)):
raise TypeError("share inputs must be validated artifacts, not text")
try:
count = len(artifacts)
except TypeError as error:
raise TypeError("share inputs must be a Sequence of artifacts") from error
if count == 0 or count > 9:
raise WrongShareCount("a share set must contain between 1 and 9 artifacts")
try:
copied = tuple(artifacts[position] for position in range(count))
except IndexError as error:
raise TypeError("share input Sequence changed length while being read") from error
if not all(isinstance(item, _Artifact) for item in copied):
raise TypeError("all share inputs must be validated codex32 artifacts")
return copied


def _artifact_tail(artifact: Share | Secret) -> tuple[tuple[int, ...], int, int]:
hrp, encoded, checksum = _decode_codex32(artifact.text)
if hrp != artifact.hrp:
raise InvalidShareSet("artifact text and validated application prefix disagree")
return tuple(encoded[6:]), checksum.length, len(encoded)
def _artifact_tail(artifact: Share | Secret) -> tuple[int, ...]:
"""Return the payload and checksum symbols of already-validated text."""
return tuple(_chars_to_u5(artifact.text[len(artifact.hrp) + 7 :].lower()))


def _validate_share_set(artifacts: Sequence[Share | Secret], *, require_exact: bool = True) -> _ShareSet:
copied = _bounded_artifacts(artifacts)
copied = tuple(islice(artifacts, 10)) # enough to reject an oversized set
if not all(isinstance(item, _Artifact) for item in copied):
raise TypeError("all share inputs must be validated codex32 artifacts")
if not 1 <= len(copied) <= 9:
raise WrongShareCount("a share set must contain between 1 and 9 artifacts")
first = copied[0]
threshold = first.header.threshold
if threshold not in range(2, 10):
raise MismatchedThreshold("Only strings with a threshold of 2 through 9 can be combined.")
if len(copied) > threshold or (require_exact and len(copied) != threshold):
raise WrongShareCount(f"threshold is {threshold}, but {len(copied)} artifacts were supplied")
first_tail, checksum_length, encoded_length = _artifact_tail(first)
tails = [first_tail]
tails = [_artifact_tail(first)]
indices = [first.header.index]
for item in copied[1:]:
if item.hrp != first.hrp:
Expand All @@ -307,15 +285,9 @@ def _validate_share_set(artifacts: Sequence[Share | Secret], *, require_exact: b
raise MismatchedThreshold("These strings require different numbers of shares.")
if item.header.identifier != first.header.identifier:
raise MismatchedIdentifier("These strings have different identifiers.")
tail, item_checksum_length, item_encoded_length = _artifact_tail(item)
item_shape = (
len(item.payload_symbols),
item_checksum_length,
item_encoded_length,
)
if item_shape != (len(first.payload_symbols), checksum_length, encoded_length):
if len(item.text) != len(first.text):
raise MismatchedPayloadLength("These strings have different lengths.")
tails.append(tail)
tails.append(_artifact_tail(item))
indices.append(item.header.index)
if len(set(indices)) != len(indices):
raise DuplicateShareIndex("That share index was already entered.")
Expand Down
2 changes: 1 addition & 1 deletion tests/test_public_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ def test_share_has_symbols_but_no_byte_or_padding_api() -> None:

def test_artifacts_cannot_be_directly_constructed_or_mutated() -> None:
with pytest.raises(TypeError):
Share("ms", Header(2, "test", "a"), (), ()) # type: ignore[arg-type]
Share("ms", Header(2, "test", "a"), (), ()) # type: ignore[call-arg]
share = parse_codex32(VECTOR_2["share_A"])
with pytest.raises((FrozenInstanceError, AttributeError)):
share.text = "changed" # type: ignore[misc]
Expand Down
14 changes: 2 additions & 12 deletions tests/test_sharing.py
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,8 @@ def test_every_threshold_recovers_exactly_k_shares(threshold: int) -> None:
extra = derive_share([secret, *shares[:-1]], IDX_SORT[threshold + 1])
with pytest.raises(WrongShareCount):
recover_secret([*shares, extra])
with pytest.raises(WrongShareCount):
recover_secret(iter(lambda: shares[0], None)) # unbounded input is rejected, not read forever


def test_share_set_mismatches_have_distinct_errors() -> None:
Expand Down Expand Up @@ -137,18 +139,6 @@ def test_share_set_mismatches_have_distinct_errors() -> None:
recover_secret(VECTOR_2["share_A"]) # type: ignore[arg-type]


def test_oversized_sequence_is_rejected_before_items_are_read() -> None:
class OversizedSequence:
def __len__(self) -> int:
return 10

def __getitem__(self, _position: int) -> Share:
raise AssertionError("oversized input must not be materialized")

with pytest.raises(WrongShareCount):
recover_secret(OversizedSequence()) # type: ignore[arg-type]


@pytest.mark.parametrize("target", ("s", "i", "b", "?", "aa", "", 3))
def test_invalid_targets_are_rejected(target: object) -> None:
secret, masks = _ms_basis()
Expand Down
Loading