diff --git a/docs/security/model.md b/docs/security/model.md index 7a7de88..2cd177f 100644 --- a/docs/security/model.md +++ b/docs/security/model.md @@ -199,10 +199,12 @@ before printing any candidate text, metadata, fingerprint, or residue addends. It then prints a conspicuous warning covering both deliberate completion of newly transcribed data and recovery with many missing characters. Literal uppercase `YES` is required before disclosure; other case variants, blank input, -or EOF terminate the command with status 1. Redirected damaged data may still -reach this gate, but disclosure requires an interactive terminal channel. If no -such channel is available, the sole message is `codex32: interactive confirmation -required` (or `ms32:`). Output formatting and `--plain` cannot bypass the gate. +or EOF terminate standalone `correct` with status 3 and correction embedded in +another workflow with status 1. Redirected damaged data may still reach this +gate, but disclosure requires an interactive terminal channel. If no such +channel is available, the sole message is `codex32: interactive confirmation +required` (or `ms32:`), with the same command-specific status. Output formatting +and `--plain` cannot bypass the gate. Existing whole-card `[y/N]` acceptance remains required after disclosure when a workflow will consume the corrected artifact. `correct` only displays the suggestion, so it has no second acceptance prompt. The gate does not verify the diff --git a/docs/user/guide.md b/docs/user/guide.md index 7be6bcf..b2d78bb 100644 --- a/docs/user/guide.md +++ b/docs/user/guide.md @@ -32,6 +32,14 @@ disclosure, workflows that consume the repaired artifact ask the usual `[y/N]` whole-card confirmation. `correct` only reports a suggestion, so it does not ask that second question. A checksum cannot make weak input secure. +The `correct` exit status distinguishes outcomes for scripts: `0` means the +input is already valid, `1` means a suggestion was emitted, `2` means the +command or input syntax was invalid, and `3` means no usable suggestion was +emitted. Status `3` includes incomplete searches with no usable suggestion, +ambiguous searches, declined disclosure, and Bitcoin Core being unavailable +when `ms32 correct` needs it to rank or fingerprint a master-seed suggestion. +The generic `codex32 correct` command does not need Core. + Choose the setup that fits you: - **Recommended: dedicated online spending wallet — easiest.** A normally diff --git a/pyproject.toml b/pyproject.toml index ce4a9fc..369196a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,7 +48,6 @@ dev = [ "twine>=5,<7", ] - [tool.setuptools.packages.find] where = ["src"] diff --git a/src/codex32/_cli_input.py b/src/codex32/_cli_input.py index c6e65de..cf5d25d 100644 --- a/src/codex32/_cli_input.py +++ b/src/codex32/_cli_input.py @@ -6,10 +6,13 @@ import difflib import os import sys -from collections.abc import Callable, Iterator +from collections.abc import Callable, Iterator, Sequence +from dataclasses import replace +from functools import partial from time import monotonic from typing import Any, Literal, cast +from codex32.bech32 import interpret_mixed_case from codex32.bip93 import ( Secret, Share, @@ -19,7 +22,7 @@ parse_codex32, recover_secret, ) -from codex32.correction import CorrectionCandidate, CorrectionContext, _best +from codex32.correction import CorrectionCandidate, CorrectionContext, _best, _capture_mass from codex32.errors import ( CodexError, DuplicateShareIndex, @@ -377,15 +380,14 @@ def _case_interpretation( profiles: tuple[Profile, ...] | None, allowed: Callable[[CorrectionCandidate], bool] | None, ) -> tuple[CorrectionCandidate | None, str, str, str] | None: - """Normalize likely casing and mark contrary-case data as erasures.""" - if value.upper() == value or value.lower() == value: - return None + # Normalize likely casing and mark contrary-case data as erasures. separator = value.find("1") base_length = separator + 1 if separator >= 0 else 0 immutable_length = len(prefix) if prefix and value.lower().startswith(prefix.lower()) else base_length - letters = [character for character in value[immutable_length:] if character.lower() != character.upper()] - uppercase = sum(character.isupper() for character in letters) > len(letters) / 2 - corrected = value.upper() if uppercase else value.lower() + interpretation = interpret_mixed_case(value, immutable_length) + if interpretation is None: + return None + corrected, erased, uppercase = interpretation corrected_prefix = prefix.upper() if uppercase else prefix.lower() try: artifact = _parse(corrected, profiles) @@ -397,14 +399,6 @@ def _case_interpretation( ) proposed = CorrectionCandidate(artifact, (), 1, 0, 0, None, capture_space_bits=bits) candidate = proposed if allowed is None or allowed(proposed) else None - erased = "".join( - corrected[index] - if index < immutable_length - or character.lower() == character.upper() - or character.isupper() == uppercase - else "?" - for index, character in enumerate(value) - ) return candidate, corrected, erased, corrected_prefix @@ -460,7 +454,10 @@ def _correction_candidates( deadline: float | None = None, capture_layers: list[tuple[int, int]] | None = None, fingerprint_match: Callable[[CorrectionCandidate], bool | None] | None = None, -) -> tuple[tuple[CorrectionCandidate, ...], bool, float | None, bool]: + seed_candidates: Sequence[CorrectionCandidate] = (), + required_only: bool = False, + optional_only: bool = False, +) -> tuple[tuple[CorrectionCandidate, ...], bool, float]: count = len(value.replace(" ", "")) targets, primary, reduced, _timed = _correction_plan(profile, byte_length, count, target) deadline = monotonic() + 10 if deadline is None else deadline @@ -476,6 +473,9 @@ def _correction_candidates( competitors=True, allowed=allowed, capture_layers=capture_layers, + seed_candidates=seed_candidates, + required_only=required_only, + optional_only=optional_only, ) if allowed is not None: candidates = tuple(candidate for candidate in candidates if allowed(candidate)) @@ -486,7 +486,66 @@ def _correction_candidates( if len(candidates) == 1 and not candidates[0].search_complete else () ) - return results, complete, deadline, False + return results, complete, deadline + + +def _scheduled_candidates( + value: str, + erased: str, + profile: str | Profile, + byte_length: int | Literal["?"] | None, + immutable: str, + excluded: tuple[str, ...] = (), + *, + target: int | None = None, + allowed: Callable[[CorrectionCandidate], bool] | None = None, + deadline: float | None = None, + fingerprint_match: Callable[[CorrectionCandidate], bool | None] | None = None, +) -> tuple[tuple[CorrectionCandidate, ...], bool]: + """Search both case interpretations under one deadline and capture ledger.""" + search = partial( + _correction_candidates, + profile=profile, + byte_length=byte_length, + immutable=immutable, + excluded=excluded, + target=target, + allowed=allowed, + fingerprint_match=fingerprint_match, + ) + first, retry = (erased, value) if erased != value else (value, None) + seeded: tuple[CorrectionCandidate, ...] = () + if retry is not None: + # Find required candidates for both case interpretations before either + # full search can spend the shared deadline on optional alignment. + # These discovery passes deliberately do not charge capture_layers; + # the full searches below account each admitted frontier once. + for required_value in (first, retry): + seeded, complete, deadline = search( + required_value, deadline=deadline, seed_candidates=seeded, required_only=True + ) + if not complete: + return (), False + capture_layers: list[tuple[int, int]] = [] + full_search = partial(search, capture_layers=capture_layers, optional_only=retry is not None) + candidates, complete, deadline = full_search(first, deadline=deadline, seed_candidates=seeded) + if retry is None: + return candidates, complete + retry_candidates, retry_complete, _deadline = full_search( + retry, deadline=deadline, seed_candidates=(*seeded, *candidates) + ) + complete = complete and retry_complete + annotated = [] + for item in (*candidates, *retry_candidates): + volume, bits = _capture_mass(capture_layers, item.capture_volume) + annotated.append(replace(item, cumulative_capture_volume=volume, capture_space_bits=bits)) + unique: dict[str, CorrectionCandidate] = {} + for item in _best(annotated, prefer_common=byte_length == "?", fingerprint_match=fingerprint_match): + # A copy from a completed earlier pass must not hide later truncation. + unique.setdefault( + item.artifact.text.lower(), item if complete else replace(item, search_complete=False) + ) + return tuple(unique.values()), complete def _fingerprint_matcher( @@ -538,23 +597,8 @@ def _suggestions( if prefix and value.lower().startswith(prefix.lower()) else prefix or value[: separator + 1] ) - deadline = monotonic() + 10 - capture_layers: list[tuple[int, int]] = [] - candidates = _correction_candidates( + return _scheduled_candidates( value, - hrp, - None, - immutable, - excluded, - target=target, - allowed=allowed, - deadline=deadline, - capture_layers=capture_layers, - fingerprint_match=fingerprint_match, - )[0] - if candidates or erased == value: - return candidates - return _correction_candidates( erased, hrp, None, @@ -562,8 +606,7 @@ def _suggestions( excluded, target=target, allowed=allowed, - deadline=deadline, - capture_layers=capture_layers, + deadline=monotonic() + 10, fingerprint_match=fingerprint_match, )[0] diff --git a/src/codex32/_competitors.py b/src/codex32/_competitors.py index 7cf5539..b5895a9 100644 --- a/src/codex32/_competitors.py +++ b/src/codex32/_competitors.py @@ -173,18 +173,25 @@ def _search_competitors( frontier: dict[_Layer, int], deadline: float, allowed: Callable[[CorrectionCandidate], bool] | None, + *, + seed_candidates: Sequence[CorrectionCandidate] = (), + optional_only: bool = False, ) -> tuple[tuple[CorrectionCandidate, ...], bool]: - results: dict[str, CorrectionCandidate] = {} + results = {candidate.artifact.text.lower(): candidate for candidate in seed_candidates} fixed: dict[int, CorrectionCandidate | None] = {} completed: set[_Layer] = set() try: for state in states: - if _FIXED in state.counts: + if not optional_only and _FIXED in state.counts: _check_deadline(deadline) fixed[state.target] = _search_fixed(state, frontier, results, allowed) targets = {state.target: state for state in states} layers = sorted( - (key for key in frontier if key[1] != _FIXED), + ( + key + for key in frontier + if key[1] != _FIXED and (not optional_only or key[1].unit != 4 and key[1].distance > 2) + ), key=lambda key: (_tier(key[1]), frontier[key]), ) for key in layers: diff --git a/src/codex32/bech32.py b/src/codex32/bech32.py index d36cf04..59befc7 100644 --- a/src/codex32/bech32.py +++ b/src/codex32/bech32.py @@ -53,6 +53,22 @@ def _validate_single_case_ascii(value: str) -> bool: return value.isupper() +def interpret_mixed_case(value: str, immutable_length: int) -> tuple[str, str, bool] | None: + # Return majority-cased and minority-erased interpretations of mixed-case text. + if not value.isascii() or value.upper() == value or value.lower() == value: + return None + letters = [character for character in value[immutable_length:] if character.isalpha()] + uppercase = sum(character.isupper() for character in letters) > len(letters) / 2 + normalized = value.upper() if uppercase else value.lower() + erased = "".join( + normalized[index] + if index < immutable_length or not character.isalpha() or character.isupper() == uppercase + else "?" + for index, character in enumerate(value) + ) + return normalized, erased, uppercase + + def bech32_encode(hrp: str, data: list[int], spec: _Checksum) -> str: """Compute a Bech32 string given HRP and data values.""" checksum = spec.create(bech32_hrp_expand(hrp) + list(data)) diff --git a/src/codex32/cli.py b/src/codex32/cli.py index 585a6dc..b20bddd 100644 --- a/src/codex32/cli.py +++ b/src/codex32/cli.py @@ -13,12 +13,13 @@ CorrectionDeclined, InteractiveConfirmationRequired, _card_text, + _case_interpretation, _confirm_correction, - _correction_candidates, _entered_groups, _fingerprint_matcher, _render_groups, _require_correction_confirmation, + _scheduled_candidates, _suggestions, ) from codex32._cli_input import InputError as _UsageError @@ -35,7 +36,12 @@ parse_codex32, recover_secret, ) -from codex32.correction import _best, _residue_low_discrimination, correct_worksheet_residue +from codex32.correction import ( + CorrectionCandidate, + _best, + _residue_low_discrimination, + correct_worksheet_residue, +) from codex32.errors import CodexError, HeaderCollision, InvalidCorrectionInput from codex32.generation import ( ConfirmationResult, @@ -420,8 +426,6 @@ def _create( raise _UsageError("--bytes applies only to a new random seed.") if not (sys.stdin.isatty() and sys.stdout.isatty()): raise _UsageError("Bitcoin backup creation requires an interactive terminal.") - if threshold and not sys.stdin.isatty(): - raise _UsageError("Shared creation requires an interactive terminal.") if threshold and shares is None and indices is None: if threshold in (2, 3): shares = {2: 3, 3: 5}[threshold] @@ -531,14 +535,16 @@ def _correct( f"Add {correction.addend} at position " f"{correction.reverse_index + 1}, counting backward from the end." ) - return 0 + return 1 if result else 0 if erasures: raise _UsageError("--erasure can be used only with --residue.") normalized = "".join(value.split()) separator = normalized.lower().rfind("1") if separator <= 0: raise _UsageError("Enter a complete application prefix followed by the separator 1.") - hrp = normalized[:separator].lower() + if len(raw_hrp := normalized[:separator]) > 83 or not all("!" <= c <= "~" for c in raw_hrp): + raise _UsageError("The application prefix must be at most 83 printable ASCII characters.") + hrp = raw_hrp.lower() if context.master_seed and hrp != Profile.MS.value: raise _UsageError("This command accepts only Bitcoin master-seed input beginning with ms1.") try: @@ -550,16 +556,25 @@ def _correct( raise _UsageError("--bytes does not match the valid master-seed backup length.") _print("The codex32 string is already valid.") return 0 - candidates, complete, _deadline, ambiguous = _correction_candidates( - value, - hrp, - byte_length, - value[: separator + 1], - ) + search_value, erased, immutable = normalized, normalized, normalized[: separator + 1] + interpreted = _case_interpretation(normalized, immutable, context.profiles, None) + if interpreted is not None: + candidate, search_value, erased, immutable = interpreted + if ( + candidate is not None + and isinstance(byte_length, int) + and len(candidate.artifact.text) != _ms_text_length(byte_length) + ): + raise _UsageError("--bytes does not match the corrected master-seed backup length.") + else: + candidate = None + if candidate is not None: + candidates: tuple[CorrectionCandidate, ...] = (candidate,) + complete = True + else: + candidates, complete = _scheduled_candidates(search_value, erased, hrp, byte_length, immutable) if not complete and not candidates: raise _CommandError("The correction search did not complete within ten seconds.") - if ambiguous: - raise _CommandError("More than one correction is possible; none was selected.") if not candidates: raise _CommandError("No valid correction found. Check the original backup.") if context.master_seed and len(candidates) > 1: @@ -679,22 +694,20 @@ def _main(context: _CliContext, argv: Sequence[str] | None = None) -> int: except SystemExit as error: return error.code if isinstance(error.code, int) else 1 scope = f"{context.prog} {arguments.command}" + correction_failed = 3 if arguments.command == "correct" else 1 try: return _dispatch(arguments, context) except CorrectionDeclined: - return 1 + return correction_failed except InteractiveConfirmationRequired: _print(f"{context.prog}: interactive confirmation required", err=True) - return 1 + return correction_failed except _UsageError as error: _print(f"{scope}: {error}", err=True) return 2 - except (_CommandError, CodexError) as error: - _print(f"{scope}: {error}", err=True) - return 1 - except BitcoinCoreError as error: + except (_CommandError, CodexError, BitcoinCoreError) as error: _print(f"{scope}: {error}", err=True) - return 1 + return correction_failed except EOFError: _print(f"{scope}: Input ended before recovery completed.", err=True) return 2 diff --git a/src/codex32/correction.py b/src/codex32/correction.py index c30ce7c..9abd345 100644 --- a/src/codex32/correction.py +++ b/src/codex32/correction.py @@ -34,6 +34,7 @@ _u5_to_chars, _validate_single_case_ascii, bech32_hrp_expand, + interpret_mixed_case, ) from codex32.bip93 import ( IDX_SORT, @@ -839,6 +840,24 @@ def _primary( ) +def _restore_case_edits( + candidates: tuple[CorrectionCandidate, ...], *, uppercase: bool +) -> tuple[CorrectionCandidate, ...]: + # Hide erasures synthesized only to search minority-case symbols. + def restore(edit: CorrectionEdit) -> CorrectionEdit: + minority_case = edit.observed.isalpha() and edit.observed.isupper() != uppercase + kind = ( + "substitution" + if edit.kind == "erasure" and minority_case and edit.observed.lower() in CHARSET + else edit.kind + ) + return replace(edit, kind=kind) if kind != edit.kind else edit + + return tuple( + replace(candidate, edits=tuple(restore(edit) for edit in candidate.edits)) for candidate in candidates + ) + + def _best( candidates: Sequence[CorrectionCandidate], *, @@ -874,30 +893,103 @@ def _correct_complete( # displayed strings are no longer than the largest expanded codeword. if len(damaged_text) > 2 * (_LONG_SPEC.period + 8): return (), True - from codex32.indel import _search_many - deadline = monotonic() + 10 if deadline is None else deadline - contexts: tuple[CorrectionContext, ...] - if context.expected_length is not None: - contexts = (context,) + base = f"{context.hrp}1" + locked = context.immutable_prefix or base + # The search strips grouping spaces, so locate the immutable boundary in + # that same coordinate system before classifying minority-case symbols. + compacted = damaged_text.replace(" ", "") + immutable_length = len(locked) if compacted.lower().startswith(locked.lower()) else len(base) + interpretation = interpret_mixed_case(compacted, immutable_length) + inputs: tuple[tuple[CorrectionContext, str], ...] + if interpretation is None: + inputs = ((context, damaged_text),) else: - # Only lengths reachable by either disjoint family are eligible. - observed = len(damaged_text.replace(" ", "")) - contexts_list = [] - for target in sorted({observed + delta for delta in (*range(-4, 5), -8, 8)}): - candidate_context = replace(context, expected_length=target) - try: - _validate_context(candidate_context) - except InvalidCorrectionInput: - continue - contexts_list.append(candidate_context) - contexts = tuple(contexts_list) - return _search_many( - contexts, - damaged_text, - primary=frozenset(c.expected_length for c in contexts if c.expected_length is not None), - deadline=deadline, + normalized, erased, uppercase = interpretation + normalized_prefix = locked.upper() if uppercase else locked.lower() + normalized_context = replace( + context, immutable_prefix=normalized_prefix if context.immutable_prefix is not None else None + ) + # Minority-case symbols are explicit erasures, so search that stronger + # interpretation before optional alignment work on the normalized text + # can consume the shared correction deadline. + inputs = ( + ((normalized_context, erased),) + if erased == normalized + else ((normalized_context, erased), (normalized_context, normalized)) + ) + + from codex32.indel import _search_many + + capture_layers: list[tuple[int, int]] = [] + candidates: tuple[CorrectionCandidate, ...] = () + complete = True + if interpretation is not None: + # Establish both interpretations' fixed/required candidates before + # either interpretation can spend the shared deadline on optional + # alignment work. These discovery passes use a private accounting + # ledger; the full searches below account every admitted layer once. + for input_context, value in inputs: + preflight_contexts: tuple[CorrectionContext, ...] + if input_context.expected_length is not None: + preflight_contexts = (input_context,) + else: + observed = len(value.replace(" ", "")) + contexts_list = [] + for target in sorted({observed + delta for delta in (*range(-4, 5), -8, 8)}): + candidate_context = replace(input_context, expected_length=target) + try: + _validate_context(candidate_context) + except InvalidCorrectionInput: + continue + contexts_list.append(candidate_context) + preflight_contexts = tuple(contexts_list) + candidates, current_complete = _search_many( + preflight_contexts, + value, + primary=frozenset( + c.expected_length for c in preflight_contexts if c.expected_length is not None + ), + deadline=deadline, + observed_text=damaged_text, + seed_candidates=candidates, + required_only=True, + ) + if not current_complete: + return (), False + for input_context, value in inputs: + contexts: tuple[CorrectionContext, ...] + if input_context.expected_length is not None: + contexts = (input_context,) + else: + # Only lengths reachable by either disjoint family are eligible. + observed = len(value.replace(" ", "")) + contexts_list = [] + for target in sorted({observed + delta for delta in (*range(-4, 5), -8, 8)}): + candidate_context = replace(input_context, expected_length=target) + try: + _validate_context(candidate_context) + except InvalidCorrectionInput: + continue + contexts_list.append(candidate_context) + contexts = tuple(contexts_list) + candidates, current_complete = _search_many( + contexts, + value, + primary=frozenset(c.expected_length for c in contexts if c.expected_length is not None), + deadline=deadline, + capture_layers=capture_layers, + observed_text=damaged_text, + seed_candidates=candidates, + optional_only=interpretation is not None, + ) + complete &= current_complete + if not current_complete and not candidates: + return (), False + candidates = ( + _restore_case_edits(candidates, uppercase=uppercase) if interpretation is not None else candidates ) + return candidates, complete def correct(context: CorrectionContext, damaged_text: str) -> tuple[CorrectionCandidate, ...]: diff --git a/src/codex32/indel.py b/src/codex32/indel.py index 9561321..26c0071 100644 --- a/src/codex32/indel.py +++ b/src/codex32/indel.py @@ -88,8 +88,6 @@ def _classes(unit: int, depth: int) -> tuple[_StructuralClass, ...]: _CHARACTER_CLASSES = _classes(1, 4) _GROUP_CLASSES = _classes(4, 2) _CLASSES = (_FIXED, *_CHARACTER_CLASSES, *_GROUP_CLASSES) -# Compatibility for older offline tools; public search uses the full class set. -_REDUCED_CLASSES = tuple(shape for shape in _CLASSES if shape.unit == 4 or shape.distance <= 3) def _group_boundary(immutable_length: int) -> int: @@ -161,44 +159,6 @@ def _alignment_counts( return counts -def _alignment_count( - shape: _StructuralClass, - observed_length: int, - target_length: int, - immutable_length: int, -) -> int: - return sum(_alignment_counts(shape, "q" * observed_length, target_length, immutable_length).values()) - - -def _reductions( - values: tuple[int, ...], - characters: str, - count: int, - offset: int, -) -> Iterator[tuple[tuple[int, ...], tuple[tuple[int, str], ...]]]: - for deleted in combinations(range(len(values)), count): - removed = frozenset(deleted) - kept = tuple(index for index in range(len(values)) if index not in removed) - retained = tuple(values[index] for index in kept) - position = 0 - for kept_index in kept: - while values[position] != values[kept_index]: - position += 1 - if position != kept_index: - break - position += 1 - else: - edits = tuple((offset + index, characters[index]) for index in deleted) - yield retained, edits - - -def _variants( - text: str, target: int, shape: _StructuralClass, immutable: int, prefix_length: int = 3 -) -> Iterator[_Variant]: - for view in _views(text, target, shape, immutable, prefix_length): - yield _view_variant(view, text, prefix_length) - - def _views(text: str, target: int, shape: _StructuralClass, immutable: int, base: int) -> Iterator[_View]: source = tuple(CHARSET.find(char.lower()) for char in text[base:]) initial = _View(source, ((0, len(source)),), len(source)) @@ -284,6 +244,7 @@ def _capacities(erasures: int, _degree: int) -> range: class _Target: context: CorrectionContext text: str + observed_text: str immutable: int target: int base: int @@ -292,12 +253,18 @@ class _Target: def _prepare( - context: CorrectionContext, damaged_text: str, classes: Sequence[_StructuralClass] + context: CorrectionContext, + damaged_text: str, + classes: Sequence[_StructuralClass], + observed_text: str | None = None, ) -> _Target | None: normalized = _normalize(context, damaged_text) if normalized is None: return None text, immutable = normalized + observed = text if observed_text is None else observed_text.replace(" ", "") + if len(observed) != len(text): + raise ValueError("observed text must preserve the searched text length") target = context.expected_length assert target is not None shapes = tuple(shape for shape in classes if shape.delta == len(text) - target) @@ -311,7 +278,31 @@ def _prepare( } base = len(context.hrp) + 1 degree = _checksum_for_encoded_length(context.hrp, target - base).length - return _Target(context, text, immutable, target, base, degree, counts) + return _Target(context, text, observed, immutable, target, base, degree, counts) + + +def _source(candidate: CorrectionCandidate, state: _Target, view: _View | None = None) -> CorrectionCandidate: + # Restore diagnostic characters transformed only to make mixed-case text searchable. + + def source_position(position: int) -> int | None: + if view is None: + return position + offset = 0 + for start, size in view.spans: + if position < offset + size: + return None if start < 0 else start + position - offset + offset += size + return None + + restored = [] + body_length = state.target - state.base + for edit in candidate.edits: + position = body_length - edit.reverse_index - 1 + source = source_position(position) + if edit.observed and source is not None and 0 <= source < len(state.observed_text) - state.base: + edit = replace(edit, observed=state.observed_text[state.base + source]) + restored.append(edit) + return replace(candidate, edits=tuple(restored)) def _layers( @@ -443,6 +434,8 @@ def _search_fixed( suspected_profile=state.context.hrp, immutable_prefix=state.context.immutable_prefix, ) + if fixed is not None: + fixed = _source(fixed, state) if fixed is None or not _allowed(state.context, fixed) or allowed is not None and not allowed(fixed): return None substitutions = sum(edit.kind == "substitution" for edit in fixed.edits) @@ -508,10 +501,12 @@ def _search_target( erasures = tuple(sorted(len(view) - p - 1 for p, _ in unknown)) fixed = solver.correct( view, - tuple((p, text[state.base + source]) for p, source in unknown if source >= 0), + tuple((p, state.observed_text[state.base + source]) for p, source in unknown if source >= 0), erasures, incremental.packed(view), ) + if fixed is not None: + fixed = _source(fixed, state, view) if fixed is None or not _allowed(context, fixed) or allowed is not None and not allowed(fixed): continue substitutions = sum(edit.kind == "substitution" for edit in fixed.edits) @@ -520,7 +515,7 @@ def _search_target( continue candidate = _adapt( fixed, - _view_variant(view, text, state.base), + _view_variant(view, state.observed_text, state.base), state.counts[shape][remaining], len(text), state.target, @@ -541,7 +536,7 @@ def _search_target( CorrectionEdit( "transposition", len(view) - offset - i - 1, - text[state.base + observed_position], + state.observed_text[state.base + observed_position], candidate.artifact.text[state.base + offset + i], ) ) @@ -563,6 +558,10 @@ def _search_many( competitors: bool = False, allowed: Callable[[CorrectionCandidate], bool] | None = None, capture_layers: list[tuple[int, int]] | None = None, + observed_text: str | None = None, + seed_candidates: Sequence[CorrectionCandidate] = (), + required_only: bool = False, + optional_only: bool = False, ) -> tuple[tuple[CorrectionCandidate, ...], bool]: deadline = monotonic() + 10 if deadline is None else deadline states = tuple( @@ -573,10 +572,23 @@ def _search_many( context, damaged_text, _CLASSES, + observed_text, ) ) is not None ) + if required_only: + states = tuple( + replace( + state, + counts={ + shape: values + for shape, values in state.counts.items() + if shape == _FIXED or shape.unit == 4 or shape.distance <= 2 + }, + ) + for state in states + ) frontier = _frontier(states, primary) layers_accounted = [] if capture_layers is None else capture_layers widths = {state.target: 5 * state.degree for state in states} @@ -594,12 +606,22 @@ def finish( if competitors: from codex32._competitors import _search_competitors - return finish(*_search_competitors(states, frontier, deadline, allowed)) - results: dict[str, CorrectionCandidate] = {} + result = _search_competitors( + states, + frontier, + deadline, + allowed, + seed_candidates=seed_candidates, + optional_only=optional_only, + ) + return finish(*result) + results = {candidate.artifact.text.lower(): candidate for candidate in seed_candidates} # One global admission ledger, then fixed, required, and optional work. # The minimum supported public sphere is A<=2 / G<=2; deeper cutoffs # require completed worst-case public-profile benchmark evidence. - for phase in (0, 1, 2): + # A mixed-case preflight may already have completed fixed/required work; + # optional_only skips only that duplicate work, not frontier admission. + for phase in (2,) if optional_only else (0, 1, 2): for original in states: counts = { shape: values diff --git a/tests/data/malformed_inputs.json b/tests/data/malformed_inputs.json index 7f38719..b904864 100644 --- a/tests/data/malformed_inputs.json +++ b/tests/data/malformed_inputs.json @@ -45,7 +45,6 @@ "correction": [ {"id": "damaged-prefix", "profile": "ms", "text": "cl10testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw"}, {"id": "damaged-separator", "profile": "ms", "text": "msx0testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw"}, - {"id": "mixed-case", "profile": "ms", "text": "Ms10testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw"}, {"id": "too-many-errors", "profile": "ms", "text": "ms12test5xxyxxuxxxxxxxxxpxxxxxxxxxx4nzvca9cmczlw"}, {"id": "non-printable", "profile": "ms", "text": "ms10testsxxxxxxxxxxxxxxxx\nxxxxxxxxxx4nzvca9cmczlw"} ], diff --git a/tests/test_cli.py b/tests/test_cli.py index b9fa03b..02e74c3 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -91,7 +91,6 @@ class _FakeBitcoinCore: chain: str = "main" version: int = 320000 imported: MasterSeed | None = None - private: bool | None = None account: int | None = None timestamp: int | str | None = None @@ -112,7 +111,7 @@ def initialize( timestamp: int | str = "now", ) -> str: self.imported = secret - self.private, self.account, self.timestamp = private, account, timestamp + self.account, self.timestamp = account, timestamp return "test-wallet" @@ -541,6 +540,17 @@ def test_xprv_suggests_mixed_case_input_with_symbol_errors(monkeypatch, capsys) assert captured.out.strip() == VECTOR_2["xprv"] +def test_embedded_correction_schedules_both_mixed_case_interpretations_before_alignment() -> None: + input_module = importlib.import_module("codex32._cli_input") + source = "ms10testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw" + damaged = "ms10testsxPxxxxxPxxxxxPxxxxxPxxxxxP4nzvca9cmczlw" + + candidates = input_module._suggestions(damaged, "", (Profile.MS,), []) + + assert len(candidates) == 1 + assert candidates[0].artifact.text == source + + def test_xprv_groups_the_next_prefix_after_spaced_correction(monkeypatch, capsys) -> None: input_module = importlib.import_module("codex32._cli_input") damaged = "NAME DLL4 F8JL H4E5 VDVU LDLF XU2J HDNL SM97 XVEN r" @@ -1715,7 +1725,7 @@ def test_cli_rejects_statistically_inadmissible_structural_burst() -> None: result = _invoke(["correct"], damaged) - assert result.exit_code == 1 + assert result.exit_code == 3 assert "No valid correction found" in result.stderr @@ -1726,10 +1736,19 @@ def test_cli_rejects_sixteen_consecutive_erasures_as_outside_regular_bound() -> assert len("".join(damaged.split())) == 48 assert damaged.count("?") == 16 - assert result.exit_code == 1 + assert result.exit_code == 3 assert "No valid correction found" in result.stderr +def test_correct_rejects_malformed_immutable_hrp_as_usage() -> None: + damaged = "é" + VECTOR_1["secret_s"][1:] + + result = _invoke(["correct"], damaged) + + assert result.exit_code == 2 + assert "application prefix" in result.stderr + + @pytest.mark.parametrize( ("byte_length", "options"), ((16, []), (64, []), (20, ["--bytes", "20"]), (24, ["--bytes", "?"])), @@ -1765,10 +1784,10 @@ def test_correction_bytes_rejects_an_unsupported_ms_size() -> None: def test_cli_never_accepts_an_incomplete_structural_search() -> None: original = VECTOR_1["secret_s"] damaged = original[:19] + original[20:] - with patch("codex32.cli._correction_candidates", return_value=((), False, 0.0, False)): + with patch("codex32._cli_input._correction_candidates", return_value=((), False, 0.0)): result = _invoke(["correct"], damaged) - assert result.exit_code != 0 + assert result.exit_code == 3 assert result.stdout == "" assert "did not complete" in result.stderr and original not in result.stderr @@ -1790,7 +1809,7 @@ def test_correction_options_control_lengths_deadline_and_search_envelope( with patch("codex32.indel._search_many", return_value=((), True)) as search: result = _invoke(["correct", *options], damaged) - assert result.exit_code == 1 and result.stdout == "" + assert result.exit_code == 3 and result.stdout == "" assert search.call_count == 1 contexts, observed = search.call_args.args assert observed == damaged @@ -1817,7 +1836,7 @@ def test_fixed_correction_repairs_legacy_cl_header_and_residue_reverse_positions residue = _invoke(["correct", "--residue"], "2ppjkw73qdjvc") assert fixed.exit_code == 1 and original in fixed.stderr - assert residue.exit_code == 0 + assert residue.exit_code == 1 assert "Add x at position 38, counting backward from the end." in residue.stdout @@ -1850,14 +1869,141 @@ def test_correction_infers_prefix_and_marks_invalid_data_as_erasures() -> None: bip39 = _invoke(["correct"], BIP39_12W_ZERO) assert removed.exit_code == 2 assert "Remove or correct these arguments: --prefix" in removed.stderr - assert damaged_prefix.exit_code == 1 + assert damaged_prefix.exit_code == 3 assert bip39.exit_code == 0 and "already valid" in bip39.stdout +def test_correct_suggests_the_majority_case_for_mixed_case_damage() -> None: + source = VECTOR_1["secret_s"] + position = next( + index for index, character in enumerate(source[3:], 3) if character.lower() != character.upper() + ) + mixed = source[:position] + source[position].upper() + source[position + 1 :] + + result = _invoke(["correct"], mixed) + wrong_length = _invoke(["correct", "--bytes", "32"], mixed) + + assert result.exit_code == 1 + assert source in result.stderr + assert "No valid correction found" not in result.stderr + assert wrong_length.exit_code == 2 + assert "--bytes does not match" in wrong_length.stderr + assert source not in wrong_length.stderr + + +def test_correct_grouped_mixed_case_recognizes_case_only_repair() -> None: + source = VECTOR_1["secret_s"] + positions = [ + index for index, character in enumerate(source[3:], 3) if character.lower() != character.upper() + ][:13] + mixed = "".join( + character.upper() if index in positions else character for index, character in enumerate(source) + ) + grouped = " ".join(mixed[index : index + 4] for index in range(0, len(mixed), 4)) + + result = _invoke(["correct"], grouped) + + assert result.exit_code == 1 and source in result.stderr + assert "interactive confirmation required" not in result.stderr + + +def test_correct_searches_mixed_case_erasures_before_normalized_alignment(monkeypatch) -> None: + source = VECTOR_1["secret_s"] + letter_positions = [ + index + for index, character in enumerate(source[3:], 3) + if index >= 9 and character.lower() != character.upper() + ] + positions = letter_positions[1:26:6] + damaged = "".join( + ("P" if character.lower() != "p" else "Q") if index in positions else character + for index, character in enumerate(source) + ) + searched: list[str] = [] + + def stop_after_first(value, *_args, **_kwargs): + searched.append(value) + return (), False, None + + monkeypatch.setattr("codex32._cli_input._correction_candidates", stop_after_first) + + result = _invoke(["correct"], damaged) + assert result.exit_code != 0 + assert len(searched) == 1 + assert searched[0].count("?") == len(positions) + + +def test_correct_reranks_mixed_case_erasure_and_normalized_interpretations() -> None: + source = "ms10testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw" + start = source.index("x") + positions = range(start, start + 13) + damaged = "".join( + ("P" if index == start + 6 else character.upper()) if index in positions else character + for index, character in enumerate(source) + ) + + result = _invoke(["correct"], damaged) + + assert result.exit_code == 1 + assert source in result.stderr + assert "interactive confirmation required" not in result.stderr + + +def test_correct_required_work_is_not_starved_by_erasure_alignment() -> None: + source = "ms10testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw" + damaged = "ms10TpstsxXxxxxxXxxxxxXxxxxXxxxxxXx4nzvcA9cmczlW" + + result = _invoke(["correct"], damaged) + + assert result.exit_code == 1 + assert source in result.stderr + assert "did not complete within ten seconds" not in result.stderr + assert "interactive confirmation required" not in result.stderr + + +def test_correct_accounts_retry_frontier_after_incomplete_first_search(monkeypatch) -> None: + source = "ms10testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw" + start = source.index("x") + positions = range(start, start + 13) + damaged = "".join( + ("P" if index == start + 6 else character.upper()) if index in positions else character + for index, character in enumerate(source) + ) + candidate = CorrectionCandidate( + parse_codex32(source), + (), + 1, + 0, + 0, + None, + search_complete=False, + ) + full_searches: list[str] = [] + + def incomplete_first(value, *_args, **kwargs): + if kwargs.get("required_only"): + return (), True, 0.0 + full_searches.append(value) + kwargs["capture_layers"].append((1, 5)) + if len(full_searches) == 1: + return (candidate,), False, 0.0 + return (), False, 0.0 + + monkeypatch.setattr("codex32._cli_input._correction_candidates", incomplete_first) + + result = _invoke(["correct"], damaged) + + assert len(full_searches) == 2 + assert full_searches[0].count("?") == len(positions) + assert "?" not in full_searches[1] + assert result.exit_code == 3 + assert "interactive confirmation required" in result.stderr + + def test_correction_hides_internal_candidate_reparse_failures() -> None: result = _invoke(["correct"], "ms12auxxxxxxxxxxxxxxxxxxxxxxxxxxxxxda3kr3s0s2swg") - assert result.exit_code != 0 + assert result.exit_code == 3 assert result.stdout == "" assert result.stderr.strip() in { "codex32 correct: No valid correction found. Check the original backup.", @@ -1875,7 +2021,6 @@ def test_wallet_commands_initialize_selected_master_seed_destinations() -> None: assert xprv.stderr.endswith("Keep it secret.\n\n") assert private.stdout == "" assert private_core.imported == parse_codex32(VECTOR_1["secret_s"]) - assert private_core.private is True assert "Warning: This gives Bitcoin Core the master private key, which can spend funds." in private.stderr assert "Use only the intended encrypted wallet" not in private.stderr assert "\x1b[" not in private.stderr + private.stdout @@ -2753,7 +2898,7 @@ def test_incomplete_candidate_has_no_search_warning_and_is_never_accepted_automa candidate = _correct_fixed(source, suspected_profile=Profile.MS) assert candidate is not None candidate = replace(candidate, search_complete=False) - with patch("codex32.cli._correction_candidates", return_value=((candidate,), False, 0.0, False)): + with patch("codex32._cli_input._correction_candidates", return_value=((candidate,), False, 0.0)): result = _invoke(["correct"], source[:-1] + "?") assert result.exit_code == 1 and result.stdout == "" assert "Search incomplete" not in result.stderr diff --git a/tests/test_correction_bch.py b/tests/test_correction_bch.py index 49b8800..a8edc80 100644 --- a/tests/test_correction_bch.py +++ b/tests/test_correction_bch.py @@ -1,7 +1,7 @@ """Independent BCH vectors, recovery bounds, and worksheet correction.""" import json -from dataclasses import FrozenInstanceError +from dataclasses import FrozenInstanceError, replace from pathlib import Path import pytest @@ -12,7 +12,7 @@ from test_profiles import _oracle_encode import codex32 -from codex32 import CorrectionCandidate, CorrectionContext, CorrectionEdit, Profile, correct +from codex32 import CorrectionCandidate, CorrectionContext, CorrectionEdit, Profile, correct, indel from codex32.bech32 import CHARSET from codex32.checksums import _CODEX32, _CODEX32_LONG from codex32.correction import ( @@ -264,6 +264,196 @@ def test_uppercase_input_preserves_case_and_reverse_addends() -> None: assert result.addend_hamming_weight == addend.bit_count() +@pytest.mark.parametrize("uppercase", (False, True)) +def test_public_correction_interprets_mixed_case_by_majority(uppercase: bool) -> None: + source = VECTOR_1["secret_s"].upper() if uppercase else VECTOR_1["secret_s"] + position = next( + index for index, character in enumerate(source[3:], 3) if character.lower() != character.upper() + ) + mixed = source[:position] + source[position].swapcase() + source[position + 1 :] + + result = correct(CorrectionContext(Profile.MS, expected_length=len(source)), mixed) + + assert len(result) == 1 + assert result[0].artifact.text == source + + +def test_grouped_mixed_case_keeps_the_recorded_header_immutable() -> None: + source = VECTOR_1["secret_s"].upper() + positions = (10, 16, 22, 28, 34) + damaged = "".join( + "p" if index in positions else character.lower() if index < 8 else character + for index, character in enumerate(source) + ) + grouped = " ".join(damaged[index : index + 4] for index in range(0, len(damaged), 4)) + context = CorrectionContext(Profile.MS, expected_length=len(source), immutable_prefix=source[:8].lower()) + + result = correct(context, grouped) + + assert len(result) == 1 + assert result[0].artifact.text == source + assert {edit.observed for edit in result[0].edits} == {"p"} + + +def test_public_correction_does_not_casefold_non_ascii() -> None: + source = VECTOR_1["secret_s"].upper() + damaged = source[:10] + "ß" + source[11:] + + result = correct(CorrectionContext(Profile.MS, expected_length=len(source)), damaged) + + assert result == () + + +def test_public_correction_searches_identical_case_interpretation_once() -> None: + source = VECTOR_1["secret_s"] + damaged = source[0].upper() + source[1:] + + result = correct(CorrectionContext(Profile.MS, expected_length=len(source)), damaged) + + assert len(result) == 1 + assert result[0].artifact.text == source + assert result[0].cumulative_capture_volume == 1 + + +def test_mixed_case_erasure_search_precedes_normalized_alignment(monkeypatch: pytest.MonkeyPatch) -> None: + source = VECTOR_1["secret_s"] + letter_positions = [ + index + for index, character in enumerate(source[3:], 3) + if index >= 9 and character.lower() != character.upper() + ] + positions = letter_positions[1:26:6] + damaged = "".join( + ("P" if character.lower() != "p" else "Q") if index in positions else character + for index, character in enumerate(source) + ) + searched: list[str] = [] + + def stop_after_first(_contexts, value, **_kwargs): + searched.append(value) + return (), False + + monkeypatch.setattr("codex32.indel._search_many", stop_after_first) + + assert correct(CorrectionContext(Profile.MS, expected_length=len(source)), damaged) == () + assert len(searched) == 1 + assert searched[0].count("?") == len(positions) + + +def test_mixed_case_candidate_still_searches_normalized_competitors(monkeypatch: pytest.MonkeyPatch) -> None: + source = VECTOR_1["secret_s"] + position = next( + index for index, character in enumerate(source[3:], 3) if character.lower() != character.upper() + ) + damaged = source[:position] + source[position].upper() + source[position + 1 :] + candidate = CorrectionCandidate(codex32.parse_codex32(source), (), 1, 0, 0, None) + searched: list[tuple[str, tuple[CorrectionCandidate, ...], bool, bool]] = [] + + def search(_contexts, value, **kwargs): # type: ignore[no-untyped-def] + seeded = kwargs["seed_candidates"] + searched.append( + (value, seeded, kwargs.get("required_only", False), kwargs.get("optional_only", False)) + ) + return ((candidate,), True) if len(searched) == 1 else (seeded, True) + + monkeypatch.setattr("codex32.indel._search_many", search) + + assert correct(CorrectionContext(Profile.MS, expected_length=len(source)), damaged) == (candidate,) + assert len(searched) == 4 + assert ["?" in value for value, _seeded, _required, _optional in searched] == [True, False, True, False] + assert [required for _value, _seeded, required, _optional in searched] == [True, True, False, False] + assert [optional for _value, _seeded, _required, optional in searched] == [False, False, True, True] + assert all(seeded == (candidate,) for _value, seeded, _required, _optional in searched[1:]) + + +def test_mixed_case_required_work_is_not_starved_by_erasure_alignment() -> None: + source = "ms10testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw" + damaged = "ms10TpstsxXxxxxxXxxxxxXxxxxXxxxxxXx4nzvcA9cmczlW" + + result = correct(CorrectionContext(Profile.MS, expected_length=len(source)), damaged) + + assert len(result) == 1 + assert result[0].artifact.text == source + + +def test_post_preflight_optional_expiry_preserves_seed(monkeypatch: pytest.MonkeyPatch) -> None: + candidate = replace( + CorrectionCandidate(codex32.parse_codex32(VECTOR_1["secret_s"]), (), 1, 0, 0, None), + capture_volume=1 << 200, + ) + searched = [] + + def expire(state, *_args, **_kwargs): # type: ignore[no-untyped-def] + searched.extend(state.counts) + return False + + monkeypatch.setattr(indel, "_search_target", expire) + result, complete = indel._search_many( + (CorrectionContext("ms", 48),), + VECTOR_1["secret_s"], + primary=frozenset((48,)), + seed_candidates=(candidate,), + optional_only=True, + ) + + assert searched + assert all(shape != indel._FIXED and shape.unit != 4 and shape.distance > 2 for shape in searched) + assert not complete + assert len(result) == 1 + assert result[0].artifact == candidate.artifact + assert not result[0].search_complete + + +@pytest.mark.parametrize("uppercase", (False, True)) +@pytest.mark.parametrize(("entered", "kind"), (("P", "substitution"), ("B", "erasure"))) +def test_mixed_case_correction_edits_preserve_the_entered_character( + uppercase: bool, + entered: str, + kind: str, +) -> None: + source = VECTOR_1["secret_s"].upper() if uppercase else VECTOR_1["secret_s"] + position = 3 + observed = entered.lower() if uppercase else entered + damaged = source[:position] + observed + source[position + 1 :] + + result = correct(CorrectionContext(Profile.MS, expected_length=len(source)), damaged) + + assert len(result) == 1 + assert result[0].artifact.text == source + assert tuple( + (edit.kind, edit.reverse_index, edit.observed, edit.replacement) for edit in result[0].edits + ) == ((kind, len(source) - position - 1, observed, source[position]),) + + +def test_mixed_case_does_not_reclassify_unrelated_structural_erasures() -> None: + source = "ms10testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw" + damaged = source[:8] + "qqqq" + source[12:] + mixed = damaged[:12] + damaged[12].upper() + damaged[13:] + + plain = correct(CorrectionContext(Profile.MS, expected_length=len(source)), damaged) + result = correct(CorrectionContext(Profile.MS, expected_length=len(source)), mixed) + + assert len(plain) == len(result) == 1 + assert result[0].artifact.text == source + assert tuple(edit.kind for edit in plain[0].edits) == ("erasure",) * 4 + assert tuple(edit.kind for edit in result[0].edits) == ("erasure",) * 4 + assert tuple((edit.reverse_index, edit.observed, edit.replacement) for edit in result[0].edits) == tuple( + (edit.reverse_index, edit.observed, edit.replacement) for edit in plain[0].edits + ) + + +def test_mixed_case_structural_edits_preserve_the_entered_character() -> None: + source = VECTOR_1["secret_s"] + damaged = source[:3] + "P" + source[4:20] + source[21:] + + result = correct(CorrectionContext(Profile.MS, expected_length=len(source)), damaged) + + assert len(result) == 1 + assert result[0].artifact.text == source + assert {edit.kind for edit in result[0].edits} == {"insertion", "substitution"} + assert next(edit for edit in result[0].edits if edit.kind == "substitution").observed == "P" + + def test_fixed_failures_are_fail_closed() -> None: mixed = "M" + VECTOR_1["secret_s"][1:] damaged = list(VECTOR_1["secret_s"]) diff --git a/tests/test_correction_disclosure.py b/tests/test_correction_disclosure.py index ffaa152..08592ee 100644 --- a/tests/test_correction_disclosure.py +++ b/tests/test_correction_disclosure.py @@ -60,7 +60,9 @@ def test_scheduler_annotates_from_admission_even_when_no_work_finished(monkeypat (48, indel._StructuralClass(0, 0, adjacent=2), 0, 0): rank + 1, } monkeypatch.setattr(indel, "_frontier", lambda *args: frontier) - monkeypatch.setattr("codex32._competitors._search_competitors", lambda *args: ((candidate,), complete)) + monkeypatch.setattr( + "codex32._competitors._search_competitors", lambda *args, **kwargs: ((candidate,), complete) + ) result, finished = indel._search_many( (CorrectionContext("ms", 48),), source, @@ -100,11 +102,11 @@ def test_noninteractive_gate_emits_only_operational_error(entrypoint, plain): patch.object(sys, "stdin", io.StringIO(VECTOR_1["secret_s"][:-1] + "?")), contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr), - patch.object(cli, "_correction_candidates", return_value=((_candidate(),), True, None, False)), + patch.object(_cli_input, "_correction_candidates", return_value=((_candidate(),), True, None)), ): status = entrypoint(["correct", *(["--plain"] if plain else [])]) prog = "codex32" if entrypoint is cli.main else "ms32" - assert status == 1 and stdout.getvalue() == "" + assert status == 3 and stdout.getvalue() == "" assert stderr.getvalue() == f"{prog}: interactive confirmation required\n" @@ -128,7 +130,7 @@ def respond(prompt, prefill=""): monkeypatch.setattr(_cli_input, "_suggestions", lambda *args, **kwargs: (candidate,)) monkeypatch.setattr(cli, "_suggestions", lambda *args, **kwargs: (candidate,)) monkeypatch.setattr( - cli, "_correction_candidates", lambda *args, **kwargs: ((candidate,), True, None, False) + _cli_input, "_correction_candidates", lambda *args, **kwargs: ((candidate,), True, None) ) core = _FakeBitcoinCore() monkeypatch.setattr(cli.BitcoinCore, "connect", lambda *args: core) @@ -140,7 +142,8 @@ def respond(prompt, prefill=""): contextlib.redirect_stderr(stderr), ): status = (cli.ms_main if command == "create" else cli.main)(args) - assert status == 1 and stdout.getvalue() == "" + expected_status = 3 if command == "correct" else 1 + assert status == expected_status and stdout.getvalue() == "" assert len(prompts) == 2 and core.imported is None warning = stderr.getvalue() assert "\x1b[1;31mWarning:\x1b[0m If you are generating new data" in warning @@ -163,7 +166,7 @@ def respond(prompt, prefill=""): monkeypatch.setattr(_cli_input, "_editable_input", respond) monkeypatch.setattr( - cli, "_correction_candidates", lambda *args, **kwargs: ((candidate,), True, None, False) + _cli_input, "_correction_candidates", lambda *args, **kwargs: ((candidate,), True, None) ) with ( patch.object(sys, "stdin", _TTYInput()), @@ -181,7 +184,7 @@ def test_redirected_stderr_blocks_low_discrimination_disclosure(monkeypatch): responses = iter((source[:-1] + "?",)) monkeypatch.setattr(_cli_input, "_editable_input", lambda *args, **kwargs: next(responses)) monkeypatch.setattr( - cli, "_correction_candidates", lambda *args, **kwargs: ((candidate,), True, None, False) + _cli_input, "_correction_candidates", lambda *args, **kwargs: ((candidate,), True, None) ) stdout, stderr = io.StringIO(), io.StringIO() with ( @@ -189,7 +192,7 @@ def test_redirected_stderr_blocks_low_discrimination_disclosure(monkeypatch): contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr), ): - assert cli.main(["correct", "--plain"]) == 1 + assert cli.main(["correct", "--plain"]) == 3 assert stdout.getvalue() == "" assert stderr.getvalue().strip() == "codex32: interactive confirmation required" @@ -218,7 +221,7 @@ def confirm(prompt): monkeypatch.setattr(_cli_input, "_confirmation_input", confirm) monkeypatch.setattr( - cli, "_correction_candidates", lambda *args, **kwargs: ((candidate,), True, None, False) + _cli_input, "_correction_candidates", lambda *args, **kwargs: ((candidate,), True, None) ) stdout, stderr = io.StringIO(), _TTYOutput() with ( @@ -292,7 +295,7 @@ def test_residue_completion_is_gated_but_ordinary_repair_is_not(degree): contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr), ): - assert cli.main(args) == 1 + assert cli.main(args) == 3 assert stdout.getvalue() == "" assert stderr.getvalue() == "codex32: interactive confirmation required\n" @@ -308,7 +311,9 @@ def test_residue_exactly_five_bits_is_not_gated_even_with_zero_addends(residue): def test_previous_case_interpretation_search_is_charged_even_without_a_candidate(monkeypatch): candidate = replace(_candidate(), capture_volume=(1 << 60) + 1) monkeypatch.setattr(indel, "_frontier", lambda *args: {(48, indel._FIXED, 0, 0): 1}) - monkeypatch.setattr("codex32._competitors._search_competitors", lambda *args: ((candidate,), True)) + monkeypatch.setattr( + "codex32._competitors._search_competitors", lambda *args, **kwargs: ((candidate,), True) + ) previous = [(1 << 60, 65)] result, _ = indel._search_many( (CorrectionContext("ms", 48),), @@ -321,6 +326,21 @@ def test_previous_case_interpretation_search_is_charged_even_without_a_candidate assert result[0].low_checksum_discrimination +def test_seed_candidate_is_reannotated_after_later_search_admission(monkeypatch): + candidate = replace(_candidate(), capture_volume=10) + monkeypatch.setattr(indel, "_frontier", lambda *args: {(48, indel._FIXED, 0, 0): 10}) + monkeypatch.setattr(indel, "_search_target", lambda *args: True) + result, complete = indel._search_many( + (CorrectionContext("ms", 48),), + VECTOR_1["secret_s"], + primary=frozenset((48,)), + capture_layers=[(5, 65)], + seed_candidates=(candidate,), + ) + assert complete and result[0].cumulative_capture_volume == 15 + assert result[0].capture_space_bits == 65 + + @pytest.mark.parametrize("profile", ("bip39_12w", "bip39_24w")) def test_cli_derives_hand_produced_bip39_set_but_rejects_invalid_implied_secret(profile): from test_generic_hrp import _invoke diff --git a/tests/test_correction_indel.py b/tests/test_correction_indel.py index 05d6d97..646e8b5 100644 --- a/tests/test_correction_indel.py +++ b/tests/test_correction_indel.py @@ -19,14 +19,11 @@ _CLASSES, _FIXED, _GROUP_CLASSES, - _REDUCED_CLASSES, - _alignment_count, _alignment_counts, _capacities, _frontier, _keep, _prepare, - _reductions, _required_header_substitutions, _search_many, _search_target, @@ -34,6 +31,7 @@ from codex32.profiles.ms32 import TEXT_LENGTHS from tools._wallet_test_vectors import FIXTURE_SEED, core_fingerprint from tools.correction_capture import cross_length_classes +from tools.correction_reference import _REDUCED_CLASSES, _alignment_count, _reductions SOURCE = VECTOR_1["secret_s"] CONTEXT = CorrectionContext(Profile.MS, expected_length=len(SOURCE)) diff --git a/tests/test_mixed_case_deadline.py b/tests/test_mixed_case_deadline.py new file mode 100644 index 0000000..f812096 --- /dev/null +++ b/tests/test_mixed_case_deadline.py @@ -0,0 +1,83 @@ +"""Deadline regression for mixed-case embedded recovery.""" + +from dataclasses import replace +from time import monotonic + +import pytest + +from codex32 import Profile +from codex32._cli_input import _case_interpretation, _correction_candidates, _scheduled_candidates + +SOURCE = "ms10testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw" + + +@pytest.mark.parametrize( + ("damaged", "erasure_fixes", "normalized_fixes"), + [ + # Five minority-case P: five erasures are correctable, five substitutions are not. + ("ms10testsxPxxxxxPxxxxxPxxxxxPxxxxxP4nzvca9cmczlw", True, False), + # Fifteen minority-case X, one mistyped: fifteen erasures are not correctable, + # one substitution after case normalization is. + ("ms10testsXXXXXXXPXXXXXXXxxxxxxxxxxx4nzvca9cmczlw", False, True), + ], +) +def test_embedded_mixed_case_recovers_either_sole_interpretation_within_deadline( + damaged: str, erasure_fixes: bool, normalized_fixes: bool +) -> None: + interpretation = _case_interpretation(damaged, "", (Profile.MS,), None) + + assert interpretation is not None + direct, normalized, erased, _prefix = interpretation + assert direct is None + immutable = normalized[: normalized.rfind("1") + 1] + for value, fixes in ((erased, erasure_fixes), (normalized, normalized_fixes)): + alone, alone_complete, _deadline = _correction_candidates( + value, Profile.MS, None, immutable, deadline=monotonic() + 10, required_only=True + ) + assert alone_complete + assert [candidate.artifact.text for candidate in alone] == ([SOURCE] if fixes else []) + + candidates, complete = _scheduled_candidates( + normalized, + erased, + Profile.MS, + None, + immutable, + deadline=monotonic() + 10, + ) + + # The exhaustive optional search may truncate at the deadline; the + # candidate must still be found and must report that truncation. + assert [candidate.artifact.text for candidate in candidates] == [SOURCE] + assert candidates[0].search_complete is complete + + +def test_scheduled_truncation_survives_a_complete_later_pass(monkeypatch: pytest.MonkeyPatch) -> None: + damaged = "ms10testsxPxxxxxPxxxxxPxxxxxPxxxxxP4nzvca9cmczlw" + interpretation = _case_interpretation(damaged, "", (Profile.MS,), None) + assert interpretation is not None + _direct, normalized, erased, _prefix = interpretation + immutable = normalized[: normalized.rfind("1") + 1] + found, _complete, _deadline = _correction_candidates( + erased, Profile.MS, None, immutable, deadline=monotonic() + 10, required_only=True + ) + full_searches: list[str] = [] + + def truncated_first(value: str, *_args: object, **kwargs: object) -> tuple[object, bool, float]: + if kwargs.get("required_only"): + return (), True, 0.0 + assert kwargs.get("optional_only") is True + full_searches.append(value) + if len(full_searches) == 1: + return (replace(found[0], search_complete=False),), False, 0.0 + return (), True, 0.0 + + monkeypatch.setattr("codex32._cli_input._correction_candidates", truncated_first) + + candidates, complete = _scheduled_candidates(normalized, erased, Profile.MS, None, immutable) + + assert full_searches == [erased, normalized] + assert not complete + assert [(candidate.artifact.text, candidate.search_complete) for candidate in candidates] == [ + (SOURCE, False) + ] diff --git a/tools/correction_benchmark.py b/tools/correction_benchmark.py index f0e74eb..922632d 100644 --- a/tools/correction_benchmark.py +++ b/tools/correction_benchmark.py @@ -15,13 +15,12 @@ from codex32.correction import CorrectionContext from codex32.indel import ( _CHARACTER_CLASSES, - _alignment_count, _required_header_substitutions, _search_many, - _variants, ) from codex32.profiles import Profile from codex32.profiles.ms32 import TEXT_LENGTHS +from tools.correction_reference import _alignment_count, _variants DELTAS = (-8, -4, -3, -2, -1, 0, 1, 2, 3, 4, 8) AUTOMATIC_48_COUNTS = (40, *range(44, 53), 56) diff --git a/tools/correction_reference.py b/tools/correction_reference.py new file mode 100644 index 0000000..b55bd40 --- /dev/null +++ b/tools/correction_reference.py @@ -0,0 +1,53 @@ +"""Reference-only structural helpers used by tests and correction benchmarks.""" + +from collections.abc import Iterator +from itertools import combinations + +from codex32.indel import ( + _CLASSES, + _alignment_counts, + _StructuralClass, + _Variant, + _view_variant, + _views, +) + +_REDUCED_CLASSES = tuple(shape for shape in _CLASSES if shape.unit == 4 or shape.distance <= 3) + + +def _alignment_count( + shape: _StructuralClass, + observed_length: int, + target_length: int, + immutable_length: int, +) -> int: + return sum(_alignment_counts(shape, "q" * observed_length, target_length, immutable_length).values()) + + +def _reductions( + values: tuple[int, ...], + characters: str, + count: int, + offset: int, +) -> Iterator[tuple[tuple[int, ...], tuple[tuple[int, str], ...]]]: + for deleted in combinations(range(len(values)), count): + removed = frozenset(deleted) + kept = tuple(index for index in range(len(values)) if index not in removed) + retained = tuple(values[index] for index in kept) + position = 0 + for kept_index in kept: + while values[position] != values[kept_index]: + position += 1 + if position != kept_index: + break + position += 1 + else: + edits = tuple((offset + index, characters[index]) for index in deleted) + yield retained, edits + + +def _variants( + text: str, target: int, shape: _StructuralClass, immutable: int, prefix_length: int = 3 +) -> Iterator[_Variant]: + for view in _views(text, target, shape, immutable, prefix_length): + yield _view_variant(view, text, prefix_length)