Skip to content
Merged
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
10 changes: 6 additions & 4 deletions docs/security/model.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
8 changes: 8 additions & 0 deletions docs/user/guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 0 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,6 @@ dev = [
"twine>=5,<7",
]


[tool.setuptools.packages.find]
where = ["src"]

Expand Down
115 changes: 79 additions & 36 deletions src/codex32/_cli_input.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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
Comment thread
BenWestgate marked this conversation as resolved.
corrected_prefix = prefix.upper() if uppercase else prefix.lower()
try:
artifact = _parse(corrected, profiles)
Expand All @@ -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


Expand Down Expand Up @@ -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
Expand All @@ -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))
Expand All @@ -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

Comment thread
BenWestgate marked this conversation as resolved.

def _fingerprint_matcher(
Expand Down Expand Up @@ -538,32 +597,16 @@ 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,
immutable,
excluded,
target=target,
allowed=allowed,
deadline=deadline,
capture_layers=capture_layers,
deadline=monotonic() + 10,
fingerprint_match=fingerprint_match,
)[0]

Expand Down
13 changes: 10 additions & 3 deletions src/codex32/_competitors.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
16 changes: 16 additions & 0 deletions src/codex32/bech32.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Comment thread
BenWestgate marked this conversation as resolved.
# 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()
Comment thread
BenWestgate marked this conversation as resolved.
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))
Expand Down
55 changes: 34 additions & 21 deletions src/codex32/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand Down Expand Up @@ -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]
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand Down Expand Up @@ -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
Expand Down
Loading
Loading