The identity key shows, imports from a file, and imports on an SPI radio

Show private key drew its warning and no key in the menu since 0.10.0.
On the uConsole and the Cardputer Zero, MeshTerm now writes an imported
key into the node's identity.key and restarts the node. An import also
reads a file (hex text or raw bytes), expands a 32-byte seed, refuses a
key of the wrong length before the confirmation, and names a firmware
that has its import turned off. The run log no longer stores the key of
config import-key (a migration removes the stored ones), and an exported
key file is written with mode 0600. Refer to #26.

Co-Authored-By: Claude Haiku 5.5 <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Haiku 5.5 committed 2026-10-12 23:51:03 -04:00
1 parent 7209d89911
commit 723f7bdfef
14 files changed
+796 -38

No files matched your search

+31
View File
@@ -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 `<secret>` 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
+3 -2
View File
@@ -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. |
+8
View File
@@ -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
+7 -3
View File
@@ -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)
+26 -3
View File
@@ -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()
+128
View File
@@ -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."
)
+32 -1
View File
@@ -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
+29 -1
View File
@@ -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 "<secret>" (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 ``<secret>``."""
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] = "<secret>"
conn.execute("UPDATE runs SET args_json = ? WHERE id = ?", (json.dumps(args), row["id"]))
+34
View File
@@ -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 ``<secret>``, 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:
"""``<secret>``, never the value."""
return "<secret>"
__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.
+47 -7
View File
@@ -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
+63 -20
View File
@@ -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)
+48
View File
@@ -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 ``<secret>`` 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("<secret>" 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",
[
+128 -1
View File
@@ -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(
+212
View File
@@ -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", "<secret>"]]},
{"ops": [["set", "name", "Lakeside"]]},
{"to": "Lakeside"},
]
assert int(version[0]) >= 19