diff --git a/CHANGELOG.md b/CHANGELOG.md index 3bca99a..2312566 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,37 @@ can change how things behave. It does not only add new things. ## [Unreleased] +### Added + +- **You can import an identity key from a file.** On Device config, **Identity key…** has + a new row, **Import from a file…**. On the command line, `config import-key --from PATH` + does the same. The file can hold the key as hex text, as `export-key --out` writes it, or + as raw bytes, as the `identity.key` of a node keeps it. +- **An import accepts a 32-byte seed.** The node that MeshTerm runs on an SPI radio keeps + its key as a seed. A firmware companion accepts only the full 64-byte key. Thus MeshTerm + makes the full key from the seed, in the same way as the firmware, before it sends it. + A key of the wrong length now stops at the prompt, before you confirm the import. + +### Fixed + +- **Show private key shows the key again.** On Device config, **Identity key…** → **Show + private key** showed its warning and no key. This occurred on each device since 0.10.0. + On the command line, `meshterm config export-key` always printed the key. +- **You can import a key on the uConsole and on the Cardputer Zero.** On these handhelds, + MeshTerm runs the node itself, and that node refused each import. Now MeshTerm writes + the key into the node's `identity.key`, and then restarts the node to use it. +- **A firmware that does not accept an imported key says so.** Some firmware builds turn + the import off. MeshTerm waited until the command timed out, and then showed a refusal + with no cause. +- **The private key no longer stays in MeshTerm's history.** MeshTerm stores the arguments + of each command that it runs. The key that you gave to `config import-key` was stored + there as plain text. Now MeshTerm stores `` in its place, and when this version + opens your database for the first time, it removes the keys that earlier versions + stored. +- **Only you can read an exported key file.** `config export-key --out` and **Export to a + file…** now write the file with mode 0600, as MeshTerm writes each other file that holds + a secret. Before, the other users of the computer could read it. + ## [0.11.0] - 2026-10-12 **MeshTerm supports the Cardputer Zero, and from a message you can open its author on the diff --git a/docs/cli/README.md b/docs/cli/README.md index b3578f7..83c6d2a 100644 --- a/docs/cli/README.md +++ b/docs/cli/README.md @@ -768,8 +768,9 @@ Show and change each device setting. If you give no subcommand, it runs `show`. | `advert [--flood]` | Broadcast an advert. It is zero-hop unless you give `--flood`. | | `share` | Print this node's contact card as a `meshcore://` URI. | | `sync-clock` | Set the device clock from this computer. | -| `export-key [--out PATH]` | Export the private key. **Sensitive.** | -| `import-key KEY_HEX --yes` | Import a private key. It overwrites this node's identity. | +| `export-key [--out PATH]` | Export the private key: 64 bytes, as 128 hex digits. **Sensitive.** Only you can read the file that `--out` writes. | +| `import-key KEY_HEX --yes` | Import a private key. It overwrites this node's identity. Give 128 hex digits, or the 64 hex digits of a 32-byte seed. | +| `import-key --from PATH --yes` | Import the key in a file: hex text, as `export-key --out` writes it, or the raw bytes of a node's `identity.key`. | | `reboot --yes` | Reboot the device. | | `factory-reset --yes` | Erase all data and reset the settings to the defaults. | diff --git a/docs/devices/uconsole.md b/docs/devices/uconsole.md index 03e258d..aac6870 100644 --- a/docs/devices/uconsole.md +++ b/docs/devices/uconsole.md @@ -222,6 +222,14 @@ in `~/.meshterm/radio/spidev1.0/`. The files are `identity.key`, `prefs.json`, MeshTerm keeps the log of the previous run as `node.log.1`. All of these files stay after a restart. +**The identity key.** `identity.key` holds the private key of the node: its identity on +the mesh. On Device config, **Identity key…** shows the key, exports it, and imports a new +one, as for each companion. An export has the form that a firmware companion uses: 64 +bytes, or 128 hex digits. When the node makes its own key, it keeps a shorter form in +`identity.key`: a 32-byte seed. The two forms are the same identity, and an import accepts +each of them. The node reads its key only when it starts. Thus, after an import, MeshTerm +restarts the node, and the reconnect dialog shows until the node is ready again. + **A profile, if you want a name for the connection:** ```toml diff --git a/meshterm/core/atomicwrite.py b/meshterm/core/atomicwrite.py index 189bbba..1a14d69 100644 --- a/meshterm/core/atomicwrite.py +++ b/meshterm/core/atomicwrite.py @@ -29,12 +29,13 @@ from pathlib import Path __all__ = ["write_atomically"] -def write_atomically(path: Path, text: str, *, owner_only: bool = False) -> None: +def write_atomically(path: Path, text: str | bytes, *, owner_only: bool = False) -> None: """Replace ``path`` with ``text`` in one step, and make parent directories if necessary. Args: path: The file to replace. Its directory is created if it does not exist. - text: The complete new contents, written as UTF-8. + text: The complete new contents. A ``str`` is written as UTF-8, and ``bytes`` are + written as they are. owner_only: Limit the file to the owner (``0600``) before it goes into position. Use it for each file that holds a credential. The function applies it to the temporary file, not to the destination. Thus the real file never exists with @@ -44,7 +45,10 @@ def write_atomically(path: Path, text: str, *, owner_only: bool = False) -> None path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(f"{path.suffix}.{os.getpid()}.tmp") try: - tmp.write_text(text, encoding="utf-8") + if isinstance(text, bytes): + tmp.write_bytes(text) + else: + tmp.write_text(text, encoding="utf-8") if owner_only: try: os.chmod(tmp, stat.S_IRUSR | stat.S_IWUSR) diff --git a/meshterm/core/connection.py b/meshterm/core/connection.py index 6cf11d9..6eb67de 100644 --- a/meshterm/core/connection.py +++ b/meshterm/core/connection.py @@ -28,7 +28,7 @@ from contextlib import AbstractAsyncContextManager, asynccontextmanager, nullcon from dataclasses import dataclass, replace from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, ClassVar from . import transmit_gate from .channels import CHANNEL_SLOT_PROBE_CAP @@ -1216,6 +1216,12 @@ class Device(ABC): events into the domain models in :mod:`meshterm.core.models`. """ + #: Whether an imported private key takes effect only when the node starts again. A + #: firmware companion uses the new key at once. The node that MeshTerm runs on an SPI + #: radio reads its key only when it starts (refer to + #: :meth:`meshterm.core.spiradio.SpiRadioDevice.import_private_key`). + import_restarts: ClassVar[bool] = False + @abstractmethod async def connect(self) -> None: """Open the connection to the device. Idempotent.""" @@ -1995,7 +2001,12 @@ class Device(ABC): @abstractmethod async def import_private_key(self, key_hex: str) -> None: - """Import a private key from a hex string (this replaces the device identity).""" + """Import a private key, and replace the identity of the device with it. + + Args: + key_hex: The 64-byte key in hex, as :func:`meshterm.core.privatekey.normalize` + makes it. A caller that has a seed expands it first. + """ @abstractmethod async def factory_reset(self) -> None: @@ -4555,7 +4566,19 @@ class MeshCoreDevice(Device): return key.hex() if isinstance(key, (bytes, bytearray)) else str(key) async def import_private_key(self, key_hex: str) -> None: # noqa: D102 - self._ok(await self._require().commands.import_private_key(bytes.fromhex(key_hex))) + from meshcore import EventType + + # The firmware has a build flag that turns the import off. Then it answers "disabled", + # and not with an error. The import command of the library waits only for "ok" or an + # error, thus it waited for its full timeout and returned a refusal with no cause. + # This call waits for the third answer too, and names it. + event = await self._require().commands.send( + b"\x18" + bytes.fromhex(key_hex), + [EventType.OK, EventType.ERROR, EventType.DISABLED], + ) + if getattr(event, "type", None) == EventType.DISABLED: + raise DeviceCommandError("this device's firmware does not accept an imported key") + self._ok(event) async def factory_reset(self) -> None: # noqa: D102 - inherited docstring mc = self._require() diff --git a/meshterm/core/privatekey.py b/meshterm/core/privatekey.py new file mode 100644 index 0000000..e4bd6e8 --- /dev/null +++ b/meshterm/core/privatekey.py @@ -0,0 +1,128 @@ +# SPDX-License-Identifier: Apache-2.0 +"""The private key of our node: the forms that MeshTerm accepts, and the one form that it sends. + +A MeshCore companion exports and imports its private key in one form only: 64 bytes, the +expanded Ed25519 key of the orlp library that the firmware uses (a 32-byte scalar, then a +32-byte nonce). In hex, this key has 128 digits. + +A key can also exist in a shorter form: the 32-byte Ed25519 seed from which the 64 bytes +come. The node that MeshTerm runs on an SPI radio keeps such a seed in its ``identity.key`` +when it makes its own key. A seed is the same identity, but a firmware companion does not +accept it. Thus MeshTerm expands a seed before it sends it (:func:`expand_seed`), in the +same way that the firmware makes its own key. The user can give either form, and the +device always gets the 64 bytes. +""" + +from __future__ import annotations + +import hashlib +import re +from pathlib import Path + +__all__ = [ + "KEY_BYTES", + "SEED_BYTES", + "expand_seed", + "normalize", + "parse_private_key", + "read_private_key_file", +] + +#: The length of a private key in the form that a companion exports and imports. +KEY_BYTES = 64 + +#: The length of an Ed25519 seed, the short form from which a key comes. +SEED_BYTES = 32 + +#: The largest file that :func:`read_private_key_file` reads. A key in hex, with a line +#: break, is 129 bytes. Thus a larger file is not a key, and the function does not read a +#: large file that the user selected by mistake. +_MAX_FILE_BYTES = 4096 + +_HEX = re.compile(r"[0-9a-fA-F]*") + + +def expand_seed(seed: bytes) -> bytes: + """The 64-byte private key that a 32-byte Ed25519 seed makes. + + This is the key generation of the orlp Ed25519 library, which the MeshCore firmware + uses: the SHA-512 digest of the seed, with its first half clamped to make the scalar. + The result has the same public key as the seed, and a firmware companion accepts it. + + Raises: + ValueError: If ``seed`` does not have 32 bytes. + """ + if len(seed) != SEED_BYTES: + raise ValueError(f"A seed has {SEED_BYTES} bytes, not {len(seed)}.") + digest = hashlib.sha512(seed).digest() + scalar = bytearray(digest[:32]) + scalar[0] &= 0xF8 + scalar[31] &= 0x3F + scalar[31] |= 0x40 + return bytes(scalar) + digest[32:] + + +def normalize(raw: bytes) -> bytes: + """``raw`` as a 64-byte key: a key stays as it is, and a seed is expanded. + + Raises: + ValueError: If ``raw`` is neither a key nor a seed. The message is for the user. + """ + if len(raw) == KEY_BYTES: + return bytes(raw) + if len(raw) == SEED_BYTES: + return expand_seed(raw) + raise ValueError( + f"A private key has {KEY_BYTES} bytes ({KEY_BYTES * 2} hex digits), or " + f"{SEED_BYTES} for a seed. This one has {len(raw)}." + ) + + +def parse_private_key(text: str) -> bytes: + """The 64-byte key in ``text``, which the user typed or pasted as hex. + + The function ignores spaces and line breaks, because a key that the user copies from a + terminal often has them where the terminal wrapped the line. + + Raises: + ValueError: If ``text`` is not hex, or has the wrong length. The message is for the + user. + """ + digits = "".join(text.split()) + if not digits or not _HEX.fullmatch(digits): + raise ValueError("A private key has hex digits only (0-9 and a-f).") + if len(digits) % 2: + raise ValueError( + f"A private key has an even number of hex digits. This one has {len(digits)}." + ) + return normalize(bytes.fromhex(digits)) + + +def read_private_key_file(path: Path) -> bytes: + """The 64-byte key in the file at ``path``: hex text, or the raw bytes of a key or a seed. + + ``config export-key --out`` writes hex text. The ``identity.key`` of a node keeps raw + bytes. The function accepts the two forms. It reads the file as hex text first. A raw + key is random bytes, and the chance that all of them are hex digits is too small to + matter. But a hex seed that someone wrote into a key file as text is a real mistake, + and as raw bytes it would be read as a different key. + + Raises: + OSError: If the file cannot be read. + ValueError: If the file does not hold a key. The message is for the user. + """ + if path.stat().st_size > _MAX_FILE_BYTES: + raise ValueError(f"{path.name} is too large to be a private key.") + data = path.read_bytes() + try: + digits = "".join(data.decode("ascii").split()) + except UnicodeDecodeError: + digits = "" + if digits and _HEX.fullmatch(digits): + return parse_private_key(digits) + if len(data) in (KEY_BYTES, SEED_BYTES): + return normalize(data) + raise ValueError( + f"{path.name} does not hold a private key. A key file holds the key as hex " + f"text, or {KEY_BYTES} or {SEED_BYTES} raw bytes." + ) diff --git a/meshterm/core/spiradio.py b/meshterm/core/spiradio.py index c5388f1..c7224a1 100644 --- a/meshterm/core/spiradio.py +++ b/meshterm/core/spiradio.py @@ -31,7 +31,8 @@ device, not of a profile. Thus, if the user gets to the same radio from the pick and from a profile the next day, it is still the same node. The first time that a radio is used, MeshTerm copies the identity and the contacts of the bridge (:func:`import_bridge_state`). Thus a node that exists already keeps its key, and with -the key its place in the contact lists of all the other nodes. +the key its place in the contact lists of all the other nodes. An imported key goes into +the same file (:meth:`SpiRadioDevice.import_private_key`). """ from __future__ import annotations @@ -48,9 +49,11 @@ from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass, replace from pathlib import Path +from .atomicwrite import write_atomically from .config import DeviceProfile, SpiWiring from .connection import DeviceCommandError, MeshCoreDevice from .discovery import TRANSPORT_SERIAL, DiscoveredDevice, spi_device +from .privatekey import KEY_BYTES, SEED_BYTES _log = logging.getLogger(__name__) @@ -879,6 +882,8 @@ class SpiRadioDevice(MeshCoreDevice): new device) starts a new node. """ + import_restarts = True + def __init__(self, wiring: SpiWiring, state: Path, *, node_name: str = "MeshTerm") -> None: """Prepare the device. Nothing starts until :meth:`connect`. @@ -932,6 +937,32 @@ class SpiRadioDevice(MeshCoreDevice): finally: await self._stop_node() + async def import_private_key(self, key_hex: str) -> None: + """Write the key into the node's ``identity.key``. The node uses it when it starts again. + + The node of the radio library refuses an import over the companion protocol. It + answers "disabled", as a firmware does when its import is off. But the key of the + node is a file that MeshTerm keeps (refer to :mod:`~meshterm.core.radionode`), and + the node reads it only when it starts. Thus this method writes the file, and does + not send the command. The running node keeps its old identity until it stops. The + caller restarts it (:attr:`~meshterm.core.connection.Device.import_restarts`): the + menu announces a reconnect, which stops this node and starts a new one, and a run + on the command line stops the node when it ends. + + The method does not restart the node itself. The services of the session subscribe + to the events of this connection, and a new connection under them would leave them + deaf. The reconnect of the session is the one path that rebuilds them. + """ + key = bytes.fromhex(key_hex) + if len(key) not in (KEY_BYTES, SEED_BYTES): + raise DeviceCommandError( + f"a private key has {KEY_BYTES} or {SEED_BYTES} bytes, not {len(key)}" + ) + await asyncio.to_thread( + write_atomically, self._state / "identity.key", key, owner_only=True + ) + _log.info("spi radio: wrote a new identity key for %s", self._wiring.spidev) + async def _stop_node(self) -> None: """Stop the node, if one runs, and forget it.""" node, self._node = self._node, None diff --git a/meshterm/persistence/db.py b/meshterm/persistence/db.py index ce0fb18..b43c9b1 100644 --- a/meshterm/persistence/db.py +++ b/meshterm/persistence/db.py @@ -8,10 +8,11 @@ new tool can store data, and the schema does not have to change again and again. from __future__ import annotations +import json import sqlite3 from pathlib import Path -SCHEMA_VERSION = 18 +SCHEMA_VERSION = 19 _SCHEMA = """ CREATE TABLE IF NOT EXISTS schema_meta ( @@ -395,3 +396,30 @@ def _migrate(conn: sqlite3.Connection) -> None: # Rows from before this column are NULL too: when MeshTerm stored such a post, the # author was lost on the way in, and nothing on disk can bring it back. conn.execute("ALTER TABLE messages ADD COLUMN author TEXT") + + stored = conn.execute("SELECT value FROM schema_meta WHERE key = 'version'").fetchone() + if stored is not None and int(stored["value"]) < 19: + # v18 -> v19: no private key in the run log. `config import-key` stored its key as + # plain text in the arguments of its run. The tools now put a secret in a wrapper + # that the run log writes as "" (refer to meshterm.tools.base.Secret). + # This step removes the keys that the earlier versions stored, one time. + _forget_imported_keys(conn) + + +def _forget_imported_keys(conn: sqlite3.Connection) -> None: + """Replace the key in each stored ``config import-key`` run with ````.""" + rows = conn.execute( + "SELECT id, args_json FROM runs WHERE tool = 'config' AND args_json LIKE '%import_key%'" + ).fetchall() + for row in rows: + try: + args = json.loads(row["args_json"]) + except ValueError: + continue + ops = args.get("ops") if isinstance(args, dict) else None + if not isinstance(ops, list): + continue + for op in ops: + if isinstance(op, list) and len(op) > 1 and op[0] == "import_key": + op[1] = "" + conn.execute("UPDATE runs SET args_json = ? WHERE id = ?", (json.dumps(args), row["id"])) diff --git a/meshterm/tools/base.py b/meshterm/tools/base.py index 5432270..75105bb 100644 --- a/meshterm/tools/base.py +++ b/meshterm/tools/base.py @@ -23,6 +23,40 @@ if TYPE_CHECKING: # do not import typer or the context at module load, for a fa from ..ui.report import Report +class Secret: + """A parameter of a run that the run log must not keep, for example a private key. + + :meth:`Tool.execute` stores the parameters of each run in the ``runs`` table, and logs + them at debug level. A secret in the parameters was thus stored as plain text: the key + of ``config import-key`` stayed in the database. A secret goes into the parameters in + this wrapper. The JSON of the run log and the log line show ````, and the tool + reads :attr:`value`. + + The class does not derive from ``str``, because ``json.dumps`` writes the value of a + ``str`` subclass and never calls its ``default`` for it. + + Attributes: + value: The secret itself. + """ + + __slots__ = ("value",) + + def __init__(self, value: str) -> None: + """Keep ``value`` out of each text form of this object.""" + self.value = value + + def __repr__(self) -> str: + """````, never the value.""" + return "" + + __str__ = __repr__ + + +def reveal(value: Any) -> Any: + """The value inside a :class:`Secret`, or ``value`` itself if it is not a secret.""" + return value.value if isinstance(value, Secret) else value + + @dataclass(slots=True) class ToolResult: """The result of a run of a tool. diff --git a/meshterm/tools/config.py b/meshterm/tools/config.py index a23b64e..06e8f62 100644 --- a/meshterm/tools/config.py +++ b/meshterm/tools/config.py @@ -19,6 +19,7 @@ from typing import TYPE_CHECKING, Any import typer from ..context import AppContext +from ..core.atomicwrite import write_atomically from ..core.channels import CHANNEL_SLOT_PROBE_CAP from ..core.config_io import backup_config, plan_restore, read_backup from ..core.connection import Device, DeviceCommandError @@ -30,8 +31,9 @@ from ..core.device_config import ( parse_value, settings_by_category, ) +from ..core.privatekey import normalize, parse_private_key, read_private_key_file from ..services.clock_sync import set_clock -from .base import Tool, ToolResult, register +from .base import Secret, Tool, ToolResult, register, reveal if TYPE_CHECKING: # pragma: no cover - typing only from ..ui.report import Block, Facts, Listing, Report @@ -170,11 +172,17 @@ class ConfigTool(Tool): @config_app.command("import-key", help="Import a private key (overwrites identity)") def _import_key_cmd( - key_hex: str = typer.Argument(..., help="Private key as hex"), + key_hex: str | None = typer.Argument( + None, help="Private key as hex: 128 digits, or a 64-digit seed" + ), + from_file: Path | None = typer.Option( + None, "--from", help="Read the key from this file (hex text or raw bytes)" + ), yes: bool = typer.Option(False, "--yes", help="Confirm this destructive action"), ) -> None: _require_yes(yes, "import-key overwrites the device identity") - run_tool_command(self, {"ops": [("import_key", key_hex)]}) + key = _private_key_argument(key_hex, from_file) + run_tool_command(self, {"ops": [("import_key", Secret(key.hex()))]}) @config_app.command("sync-clock", help="Set the device clock from this computer") def _sync_clock_cmd() -> None: @@ -312,8 +320,13 @@ async def apply_ops( elif kind == "export_key": report.append(await _export_key(ctx, device, op[1] if len(op) > 1 else None, artifacts)) elif kind == "import_key": - await device.import_private_key(op[1]) - ctx.ui.ack("[ok]✓[/ok] private key imported") + # The callers check the key before they confirm. This expands a seed that a + # caller gives as it is, because a firmware companion accepts only 64 bytes. + await device.import_private_key(normalize(bytes.fromhex(reveal(op[1]))).hex()) + if device.import_restarts: + ctx.ui.ack("[ok]✓[/ok] private key imported — the node uses it from its next start") + else: + ctx.ui.ack("[ok]✓[/ok] private key imported") changes += 1 report.append(_acted("imported", changes=1, imported=True)) elif kind == "factory_reset": @@ -635,20 +648,28 @@ async def _export_key( This document holds a private key when the caller asked for one. That is the purpose of the command, the same as on the plain face, and this function adds no more gate. + + The menu has no renderer for the report. Thus there, the function also draws the key + itself, as :func:`_share_contact` draws its code. When the report replaced the line + that printed the key, "Show private key" drew its warning and no key at all. + + The file is for the owner only (``0600``), as each other file that holds a credential. """ from ..ui import fields from ..ui.report import BARE, Facts + from ..ui.surface import TuiUi key_hex = await device.export_private_key() written: Path | None = None if out is not None: written = Path(out) - written.parent.mkdir(parents=True, exist_ok=True) - written.write_text(key_hex, encoding="utf-8") + write_atomically(written, key_hex, owner_only=True) artifacts.append(str(written)) ctx.ui.ack("[warn]private key written — keep this file secret.[/warn]") else: ctx.ui.ack("[warn]private key (keep secret):[/warn]") + if isinstance(ctx.ui, TuiUi): + ctx.ui.note(key_hex) return Facts( key="key", fields=(fields.hexid("private_key", "private_key"), fields.word("path", "path")), @@ -694,6 +715,25 @@ def _require_yes(yes: bool, what: str) -> None: raise typer.BadParameter(f"{what}. Re-run with --yes to confirm.") +def _private_key_argument(key_hex: str | None, from_file: Path | None) -> bytes: + """The 64-byte key that ``config import-key`` got, from its argument or from ``--from``. + + The command reads and checks the key before it connects. Thus a bad key sends nothing + to the device, and it is a usage error (exit 2), as each other bad argument is. + + Raises: + typer.BadParameter: If the command got no key, two keys, or a key that is not one. + """ + if (key_hex is None) == (from_file is None): + raise typer.BadParameter("give the key as hex, or --from with a file, but not both.") + try: + if from_file is not None: + return read_private_key_file(from_file) + return parse_private_key(key_hex or "") + except (OSError, ValueError) as exc: + raise typer.BadParameter(str(exc)) from None + + def _setting_columns() -> tuple: """The columns of one setting row, shared by ``config show`` and ``config get``.""" from ..ui import fields diff --git a/meshterm/ui/config_editor.py b/meshterm/ui/config_editor.py index a6bdbf4..99d3e3a 100644 --- a/meshterm/ui/config_editor.py +++ b/meshterm/ui/config_editor.py @@ -56,6 +56,7 @@ from ..core.device_config import ( parse_value, settings_by_category, ) +from ..core.privatekey import parse_private_key, read_private_key_file from ..platforms import Platform, on_platform from .device_info_screen import REVEAL_KEY from .marks import MASK_MARK @@ -212,7 +213,14 @@ async def edit_config(ctx: AppContext) -> dict[str, Any] | None: snapshot, custom = await _reread(ctx, device) _unstage_held(snapshot, custom, pending, custom_pending) elif choice == _IDENTITY_KEY: - if await _identity_key_menu(ctx, device, snapshot): + if await _identity_key_menu(ctx, device, snapshot, staged()): + if device.import_restarts and ctx.active_transport is not None: + # The node reads its new key only when it starts. The reconnect of + # the session stops this node and starts a new one, as after a + # reboot, and the visit ends with it (refer to _reboot). + ctx.announce_reboot() + await asyncio.sleep(_REBOOT_HANDOFF_TIMEOUT_S) + return summary() snapshot, custom = await _reread(ctx, device) elif choice == _REBOOT: # A reboot ends the visit, and the staged values go with it. Thus ask first. @@ -1352,12 +1360,27 @@ async def _restore_now(ctx: AppContext, device: Device, snapshot: dict) -> bool: return changed > 0 -async def _identity_key_menu(ctx: AppContext, device: Device, snapshot: dict) -> bool: +async def _identity_key_menu( + ctx: AppContext, device: Device, snapshot: dict, staged: int = 0 +) -> bool: """Export or import the private identity key of the device. + An import takes a pasted key or a file. Both accept the 64-byte key that a companion + exports, and also the 32-byte seed that the node of an SPI radio keeps + (:mod:`meshterm.core.privatekey`). Each input is checked before the confirmation, so + a key that is not one never gets to the typed confirmation. + + Args: + ctx: The shared application context. + device: The connected device. + snapshot: The current device snapshot. + staged: The number of staged changes on the page. An import that restarts the node + ends the visit, as a reboot does, so the user must agree to discard them first. + Returns: ``True`` if the identity changed (a key was imported). Then the caller reads its - snapshot again. + snapshot again, or gives control to the reconnect if the node must restart + (:attr:`~meshterm.core.connection.Device.import_restarts`). """ choice = await ctx.ui.select( "Identity key", @@ -1366,7 +1389,8 @@ async def _identity_key_menu(ctx: AppContext, device: Device, snapshot: dict) -> [ ("Show private key", "Display it on screen (sensitive)", "show"), ("Export to a file…", "Write it to disk (keep it secret)", "file"), - ("Import a key…", "Replace this device's identity", "import"), + ("Import a key…", "Paste it (replaces the identity)", "import"), + ("Import from a file…", "Read it from disk (replaces the identity)", "load"), ] ), ], @@ -1399,22 +1423,41 @@ async def _identity_key_menu(ctx: AppContext, device: Device, snapshot: dict) -> return False # Import: get the key, then ask for the typed confirmation. - key_hex = await ctx.ui.text( - "Import identity key", - prompt="Paste the private key as hex:", - validate=_is_hex, - ) - if not key_hex: + if choice == "load": + raw = await ctx.ui.path( + "Import identity key", prompt="Read the private key from this file:" + ) + if not raw: + return False + try: + key = read_private_key_file(Path(raw)) + except (OSError, ValueError) as exc: + ctx.ui.note(f"[err]not imported:[/err] {escape(str(exc))}") + await ctx.ui.present(title="Import key") + return False + else: + key_hex = await ctx.ui.text( + "Import identity key", + prompt="Paste the private key as hex:", + validate=_is_private_key, + ) + if not key_hex: + return False + key = parse_private_key(key_hex) + + restarts = device.import_restarts and ctx.active_transport is not None + if restarts and staged and not await confirm_discard(ctx, staged, verb="applying"): return False - confirmed = await ctx.ui.typed_confirm( + warning = ( "Importing a key permanently overwrites this device's identity. Contacts and " - "messages keyed to the old identity will no longer match it.", - "IMPORT", - title="Import private key", + "messages keyed to the old identity will no longer match it." ) + if restarts: + warning += " The node restarts to use the new key." + confirmed = await ctx.ui.typed_confirm(warning, "IMPORT", title="Import private key") if not confirmed: return False - await _run_now(ctx, device, snapshot, [("import_key", key_hex.strip())], "Import key") + await _run_now(ctx, device, snapshot, [("import_key", key.hex())], "Import key") return True @@ -1440,10 +1483,10 @@ async def _factory_reset(ctx: AppContext, device: Device, snapshot: dict) -> boo # --- validators -------------------------------------------------------------- -def _is_hex(text: str) -> bool | str: - """Check that ``text`` is a hex string.""" +def _is_private_key(text: str) -> bool | str: + """Check that ``text`` is a private key in hex: ``True``, or the problem for the prompt.""" try: - bytes.fromhex(text) + parse_private_key(text) return True - except ValueError: - return "Enter hex characters only." + except ValueError as exc: + return str(exc) diff --git a/tests/test_cli_contract.py b/tests/test_cli_contract.py index 45040d2..98689eb 100644 --- a/tests/test_cli_contract.py +++ b/tests/test_cli_contract.py @@ -19,7 +19,9 @@ Each command here is read-only or uses the simulator, so nothing transmits. from __future__ import annotations import json +import os import re +import stat from pathlib import Path import pytest @@ -579,6 +581,52 @@ def test_a_destructive_command_without_yes_is_a_silent_usage_error(run, args) -> assert "--yes" in result.stderr +def test_import_key_reads_a_file_and_keeps_the_key_out_of_the_run_log(run, tmp_path: Path) -> None: # noqa: ANN001 + """The run log stores the parameters of each run, and a private key is not one to keep. + + The key of ``import-key`` was stored as plain text in the ``runs`` table. Now the table + holds ```` where the key was. + """ + import sqlite3 + + key = "cd" * 64 + path = tmp_path / "node.key" + path.write_text(key + "\n", encoding="utf-8") + assert run("config", "import-key", "--from", str(path), "--yes").exit_code == exitcodes.OK + assert run("config", "import-key", key, "--yes").exit_code == exitcodes.OK + with sqlite3.connect(tmp_path / "test.db") as db: + stored = [row[0] for row in db.execute("SELECT args_json FROM runs")] + assert len(stored) == 2 + assert all("" in args and key not in args for args in stored) + + +@pytest.mark.parametrize( + "args,problem", + [ + ((), "but not both"), + (("cd" * 64, "--from", "node.key"), "but not both"), + (("not-a-key",), "hex digits only"), + (("cd" * 40,), "This one has 40"), + (("--from", "no-such-file"), "no-such-file"), + ], + ids=["no-key", "two-keys", "not-hex", "wrong-length", "no-file"], +) +def test_import_key_refuses_a_key_that_is_not_one(run, args, problem) -> None: # noqa: ANN001 + """A bad key is a bad argument: a usage error, before anything goes to the device.""" + result = run("config", "import-key", *args, "--yes") + assert result.exit_code == exitcodes.USAGE + assert result.stdout == "" + assert problem in " ".join(result.stderr.split()) + + +@pytest.mark.skipif(os.name != "posix", reason="Windows has no owner-only mode bits") +def test_export_key_writes_a_file_for_the_owner_only(run, tmp_path: Path) -> None: # noqa: ANN001 + """The file holds a private key, so nobody else on the machine can read it.""" + path = tmp_path / "node.key" + assert run("config", "export-key", "--out", str(path)).exit_code == exitcodes.OK + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + @pytest.mark.parametrize( "args,flag", [ diff --git a/tests/test_config_editor.py b/tests/test_config_editor.py index b1b242e..9c52c3d 100644 --- a/tests/test_config_editor.py +++ b/tests/test_config_editor.py @@ -22,10 +22,12 @@ from meshterm.context import AppContext from meshterm.core.admin_store import AdminStore from meshterm.core.config import Settings from meshterm.core.device_store import DeviceStore +from meshterm.core.privatekey import expand_seed from meshterm.persistence.repository import Repository from meshterm.platforms import PICOCALC_LYRA, REGULAR, set_platform from meshterm.ui.config_editor import ( _ConfigMenu, + _is_private_key, _menu_items, config_table, contact_share_url, @@ -1100,7 +1102,7 @@ async def test_actions_factory_reset_gates_on_typed_confirmation(ctx: AppContext async def test_actions_import_key_gates_on_typed_confirmation(ctx: AppContext) -> None: """An import of an identity key needs the typed IMPORT confirmation.""" device = await ctx.device() - new_key = "ab" * 32 + new_key = "ab" * 64 _install( ctx, [ @@ -1115,6 +1117,131 @@ async def test_actions_import_key_gates_on_typed_confirmation(ctx: AppContext) - assert await device.export_private_key() == new_key +async def test_actions_show_key_draws_the_key( + ctx: AppContext, monkeypatch: pytest.MonkeyPatch +) -> None: + """Show private key draws the key itself. For a time, it drew its warning and no key.""" + monkeypatch.setattr("meshterm.ui.surface.TuiUi", _ScriptedUi) + key = await (await ctx.device()).export_private_key() + ui = _install( + ctx, + [ + ("select", "__identity_key__"), + ("select", "show"), + ("dialog", "show"), + ("select", None), + ], + ) + await edit_config(ctx) + assert key in ui.notes + assert ui.presented == ["Private key"] + + +async def test_actions_import_key_expands_a_seed(ctx: AppContext) -> None: + """A pasted 32-byte seed goes to the device as the 64-byte key that a firmware accepts.""" + device = await ctx.device() + seed = "ab" * 32 + _install( + ctx, + [ + ("select", "__identity_key__"), + ("select", "import"), + ("text", seed), + ("typed_confirm", True), + ("select", None), + ], + ) + await edit_config(ctx) + assert await device.export_private_key() == expand_seed(bytes.fromhex(seed)).hex() + + +@pytest.mark.parametrize( + "content", + [("cd" * 64 + "\n").encode(), bytes.fromhex("cd" * 64)], + ids=["hex-text", "raw-bytes"], +) +async def test_actions_import_key_reads_a_file( + ctx: AppContext, tmp_path: Path, content: bytes +) -> None: + """A file from ``export-key --out`` and a raw ``identity.key`` import the same key.""" + device = await ctx.device() + path = tmp_path / "node.key" + path.write_bytes(content) + _install( + ctx, + [ + ("select", "__identity_key__"), + ("select", "load"), + ("path", str(path)), + ("typed_confirm", True), + ("select", None), + ], + ) + await edit_config(ctx) + assert await device.export_private_key() == "cd" * 64 + + +async def test_actions_import_key_stops_at_a_file_that_holds_no_key( + ctx: AppContext, tmp_path: Path +) -> None: + """A file that holds no key says why, and never gets to the confirmation.""" + device = await ctx.device() + before = await device.export_private_key() + path = tmp_path / "notes.txt" + path.write_bytes(b"\x00" * 10) + ui = _install( + ctx, + [ + ("select", "__identity_key__"), + ("select", "load"), + ("path", str(path)), + ("select", None), # no typed confirmation: the script would fail on one + ], + ) + await edit_config(ctx) + assert await device.export_private_key() == before + assert any("not imported" in note for note in ui.notes) + + +def test_a_pasted_key_of_the_wrong_length_says_why() -> None: + """The prompt refuses the key before the confirmation, and names the problem.""" + assert _is_private_key("ab" * 64) is True + assert _is_private_key("AB" * 32 + "\n") is True + assert "This one has 40" in str(_is_private_key("ab" * 40)) + assert "hex digits only" in str(_is_private_key("not a key")) + + +async def test_actions_import_key_restarts_a_node_that_reads_its_key_at_start( + ctx: AppContext, monkeypatch: pytest.MonkeyPatch +) -> None: + """On the node of an SPI radio, the import ends the visit and gives control to the reconnect. + + The staged change goes with the visit. Thus the user agrees to discard it first. + """ + from meshterm.ui import config_editor + + device = await ctx.device() + monkeypatch.setattr(type(device), "import_restarts", True) + monkeypatch.setattr(AppContext, "active_transport", property(lambda _self: "spi")) + monkeypatch.setattr(config_editor, "_REBOOT_HANDOFF_TIMEOUT_S", 0) + ui = _install( + ctx, + [ + ("select", "name"), + ("text", "NewName"), + ("select", "__identity_key__"), + ("select", "import"), + ("text", "ab" * 64), + ("dialog", "discard"), + ("typed_confirm", True), + ], + ) + assert await edit_config(ctx) is None + assert ctx.reboot_in_progress + assert not ui.script + assert await device.export_private_key() == "ab" * 64 + + async def test_actions_reboot_on_simulator_stays_on_the_screen(ctx: AppContext) -> None: """The simulator has no link to drop, so a confirmed reboot only adds a note and returns.""" ui = _install( diff --git a/tests/test_privatekey.py b/tests/test_privatekey.py new file mode 100644 index 0000000..600140f --- /dev/null +++ b/tests/test_privatekey.py @@ -0,0 +1,212 @@ +# SPDX-License-Identifier: Apache-2.0 +"""The private key of our node: the forms that MeshTerm accepts, and how each device takes it. + +A companion exports and imports a 64-byte key. A node of an SPI radio keeps a 32-byte seed. +MeshTerm accepts the two forms from the user and sends only the first form, and these tests +hold each step of that path. +""" + +from __future__ import annotations + +import os +import stat +from pathlib import Path +from types import SimpleNamespace + +import pytest +from Crypto.PublicKey import ECC +from meshcore import EventType +from meshcore.events import Event + +from meshterm.core import spiradio +from meshterm.core.config import SpiWiring +from meshterm.core.connection import DeviceCommandError, MeshCoreDevice +from meshterm.core.privatekey import ( + expand_seed, + normalize, + parse_private_key, + read_private_key_file, +) + +#: The base point of Ed25519 (RFC 8032, section 5.1). +_GX = 15112221349535400772501151409588531511454012693041857206046113283949847762202 +_GY = 46316835694926478169428394003475163141307993866256225615783033603165251855960 + + +# --- the forms ----------------------------------------------------------------------------- + + +def test_an_expanded_seed_has_the_public_key_of_the_seed() -> None: + """The firmware gets the public key from the first half of the key, with no more clamp. + + That point must be the public key that Ed25519 makes from the seed. If it is not, the + import gives the device a different identity from the one the user exported. + """ + seed = bytes(range(32)) + key = expand_seed(seed) + assert len(key) == 64 + point = ECC.EccPoint(_GX, _GY, curve="Ed25519") * int.from_bytes(key[:32], "little") + expected = ECC.construct(curve="Ed25519", seed=seed).pointQ + assert (point.x, point.y) == (expected.x, expected.y) + + +def test_an_expanded_seed_is_clamped_as_the_firmware_clamps_it() -> None: + """The three low bits are clear, the top bit is clear, and the bit below it is set.""" + key = expand_seed(b"\xff" * 32) + assert key[0] & 0x07 == 0 + assert key[31] & 0x80 == 0 + assert key[31] & 0x40 + + +def test_a_key_passes_as_it_is_and_a_seed_is_expanded() -> None: + """The device always gets 64 bytes.""" + key = bytes.fromhex("cd" * 64) + assert normalize(key) == key + assert normalize(key[:32]) == expand_seed(key[:32]) + + +def test_a_pasted_key_can_have_line_breaks_and_capitals() -> None: + """A key that the user copies from a terminal has the breaks where the terminal wrapped it.""" + pasted = "CD" * 32 + "\n " + "cd" * 32 + "\n" + assert parse_private_key(pasted) == bytes.fromhex("cd" * 64) + + +@pytest.mark.parametrize( + "text,problem", + [ + ("", "hex digits only"), + ("not a key", "hex digits only"), + ("abc", "even number"), + ("ab" * 40, "This one has 40"), + ], + ids=["empty", "not-hex", "odd", "wrong-length"], +) +def test_a_bad_key_says_why(text: str, problem: str) -> None: + """The message is for the user, who must correct what they pasted.""" + with pytest.raises(ValueError, match=problem): + parse_private_key(text) + + +@pytest.mark.parametrize( + "content,expected", + [ + (("cd" * 64 + "\n").encode(), bytes.fromhex("cd" * 64)), + (bytes.fromhex("cd" * 64), bytes.fromhex("cd" * 64)), + (bytes(range(32)), expand_seed(bytes(range(32)))), + (("ab" * 32).encode(), expand_seed(bytes.fromhex("ab" * 32))), + ], + ids=["hex-text", "raw-key", "raw-seed", "seed-as-hex-text"], +) +def test_a_key_file_can_hold_hex_text_or_raw_bytes( + tmp_path: Path, content: bytes, expected: bytes +) -> None: + """``export-key --out`` writes hex text, and the ``identity.key`` of a node holds raw bytes. + + A seed that someone wrote into a file as hex text is 64 bytes on disk. As raw bytes, it + would be a different key with no error, so the file is read as text first. + """ + path = tmp_path / "node.key" + path.write_bytes(content) + assert read_private_key_file(path) == expected + + +def test_a_file_that_holds_no_key_says_so(tmp_path: Path) -> None: + """A file of the wrong size or contents is refused, with its name in the message.""" + path = tmp_path / "notes.bin" + path.write_bytes(b"\x00" * 10) + with pytest.raises(ValueError, match="notes.bin does not hold a private key"): + read_private_key_file(path) + path.write_bytes(b"\x00" * 5000) + with pytest.raises(ValueError, match="too large"): + read_private_key_file(path) + + +# --- the devices --------------------------------------------------------------------------- + + +def _device_that_answers(event: Event) -> tuple[MeshCoreDevice, list[bytes]]: + """A device whose client answers each command with ``event``, and the frames it got.""" + sent: list[bytes] = [] + + async def send(data: bytes, expected: list, timeout: float | None = None) -> Event: + sent.append(data) + assert event.type in expected, "the device must wait for this answer" + return event + + device = MeshCoreDevice(port="mock") + device._mc = SimpleNamespace(commands=SimpleNamespace(send=send)) + return device, sent + + +async def test_a_firmware_import_sends_the_key() -> None: + """The command is 0x18 and the 64 bytes of the key.""" + device, sent = _device_that_answers(Event(EventType.OK, {})) + await device.import_private_key("cd" * 64) + assert sent == [b"\x18" + bytes.fromhex("cd" * 64)] + + +async def test_a_firmware_with_the_import_off_says_so() -> None: + """The firmware answers "disabled". Before, MeshTerm waited for its timeout and said nothing.""" + device, _sent = _device_that_answers(Event(EventType.DISABLED, {})) + with pytest.raises(DeviceCommandError, match="does not accept an imported key"): + await device.import_private_key("cd" * 64) + + +async def test_an_spi_node_gets_its_imported_key_as_a_file(tmp_path: Path) -> None: + """The node of the radio library refuses an import, so MeshTerm writes its key file. + + The node reads the file when it starts again, so the device says that the import + restarts it. + """ + device = spiradio.SpiRadioDevice(SpiWiring(), tmp_path / "state") + assert device.import_restarts + key = expand_seed(bytes(range(32))) + await device.import_private_key(key.hex()) + path = tmp_path / "state" / "identity.key" + assert path.read_bytes() == key + if os.name == "posix": + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_a_firmware_companion_uses_its_new_key_at_once() -> None: + """Only a node that MeshTerm runs must start again to use the key.""" + assert MeshCoreDevice.import_restarts is False + + +# --- the run log --------------------------------------------------------------------------- + + +def test_a_v18_database_forgets_the_keys_that_import_key_stored(tmp_path: Path) -> None: + """Before v19, the run of ``config import-key`` stored its key. The migration removes it. + + The other runs and the other arguments stay as they were. + """ + import json + + from meshterm.persistence import db as db_module + + path = tmp_path / "old.db" + conn = db_module.connect(path) + conn.execute("UPDATE schema_meta SET value = '18' WHERE key = 'version'") + for tool, args in ( + ("config", {"ops": [["import_key", "cd" * 64]]}), + ("config", {"ops": [["set", "name", "Lakeside"]]}), + ("trace", {"to": "Lakeside"}), + ): + conn.execute( + "INSERT INTO runs (tool, args_json, started_at) VALUES (?, ?, '2026-10-01')", + (tool, json.dumps(args)), + ) + conn.commit() + conn.close() + + conn = db_module.connect(path) + stored = [json.loads(row[0]) for row in conn.execute("SELECT args_json FROM runs ORDER BY id")] + version = conn.execute("SELECT value FROM schema_meta WHERE key = 'version'").fetchone() + conn.close() + assert stored == [ + {"ops": [["import_key", ""]]}, + {"ops": [["set", "name", "Lakeside"]]}, + {"to": "Lakeside"}, + ] + assert int(version[0]) >= 19