Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
488db7f278 | ||
|
|
b434cb1643 | ||
|
|
723f7bdfef | ||
|
|
7209d89911 |
No files matched your search
@@ -184,7 +184,10 @@ gh pr list -R CardputerZero/packages --author jpmartineau --state all --limit 3
|
||||
If `store` failed, the usual causes are the `CARDPUTER_STORE_TOKEN` secret (missing, or the
|
||||
token revoked) or a change in czdev: the job pins AppBuilder to one commit because
|
||||
`scripts/cardputer-zero/store-publish.py` corrects two of its functions, and the script
|
||||
refuses a czdev without them. A rerun is safe — the script publishes nothing when the store
|
||||
refuses a czdev without them. The script also fast-forwards JP's fork of the store before
|
||||
czdev branches from it; it stops if that fails (the fork has commits of its own), because
|
||||
a stale fork makes the store's check fail on every file the store has removed since. A
|
||||
rerun is safe — the script publishes nothing when the store
|
||||
already has the version or a PR for it is open. While a PR for an **earlier** version is
|
||||
still open (a first release waits for a maintainer and a video), the job only warns: a
|
||||
second PR would add the same files and conflict. The release itself is not held back. Once
|
||||
|
||||
+43
-1
@@ -9,6 +9,47 @@ can change how things behave. It does not only add new things.
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
## [0.11.1] - 2026-10-13
|
||||
|
||||
**Show private key shows the key again, and the uConsole and the Cardputer Zero can import
|
||||
an identity key.** On each device, the menu showed the warning of Show private key and no
|
||||
key. On the handhelds that MeshTerm drives over SPI, an import of a key failed. An import
|
||||
now also reads a file and accepts a 32-byte seed. MeshTerm no longer keeps the private key
|
||||
in its history of commands.
|
||||
|
||||
### 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.
|
||||
([#26](https://github.com/jpmartineau/MeshTerm/issues/26))
|
||||
- **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.
|
||||
([#26](https://github.com/jpmartineau/MeshTerm/issues/26))
|
||||
- **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
|
||||
@@ -499,7 +540,8 @@ commands.
|
||||
MeshTerm is free and open source, under the Apache 2.0 licence. The licence does not cover
|
||||
the name and the logo (refer to `NOTICE`). A fork is welcome under its own name.
|
||||
|
||||
[Unreleased]: https://github.com/jpmartineau/MeshTerm/compare/v0.11.0...HEAD
|
||||
[Unreleased]: https://github.com/jpmartineau/MeshTerm/compare/v0.11.1...HEAD
|
||||
[0.11.1]: https://github.com/jpmartineau/MeshTerm/releases/tag/v0.11.1
|
||||
[0.11.0]: https://github.com/jpmartineau/MeshTerm/releases/tag/v0.11.0
|
||||
[0.10.3]: https://github.com/jpmartineau/MeshTerm/releases/tag/v0.10.3
|
||||
[0.10.2]: https://github.com/jpmartineau/MeshTerm/releases/tag/v0.10.2
|
||||
|
||||
+4
-3
@@ -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. |
|
||||
|
||||
@@ -1799,7 +1800,7 @@ code beside each link. The scripted face prints only the link.
|
||||
**`--json`:**
|
||||
|
||||
```json
|
||||
{"page":"discord","title":"Join Discord","version":"0.11.0","text":"Join the Discord\n Questions, ideas, bug reports, and mesh talk.\n\n • https://discord.gg/AZwe5Uvb3S","links":["https://discord.gg/AZwe5Uvb3S"]}
|
||||
{"page":"discord","title":"Join Discord","version":"0.11.1","text":"Join the Discord\n Questions, ideas, bug reports, and mesh talk.\n\n • https://discord.gg/AZwe5Uvb3S","links":["https://discord.gg/AZwe5Uvb3S"]}
|
||||
```
|
||||
|
||||
| Field | Type | Meaning |
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -6,7 +6,7 @@ Connects to a companion over USB, Bluetooth, or TCP. For Windows, macOS, and Lin
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
__version__ = "0.11.0"
|
||||
__version__ = "0.11.1"
|
||||
__author__ = "Jean-Pierre Martineau"
|
||||
|
||||
#: The year when MeshTerm was first published. The copyright span starts in this year.
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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."
|
||||
)
|
||||
@@ -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
|
||||
|
||||
@@ -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"]))
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
@@ -5,7 +5,8 @@ r"""Publish the Cardputer Zero package to M5Stack's app store, with ``czdev`` an
|
||||
``czdev publish`` (M5Stack's CardputerZero-AppBuilder) uploads the ``.deb`` to a release in
|
||||
the fork of ``CardputerZero/packages``, pushes a branch with the store listing, and opens a
|
||||
pull request. The store merges an update by itself when its checks pass. This script runs
|
||||
``czdev`` in its own process with two corrections, and it sets up ``git`` for it:
|
||||
``czdev`` in its own process with two corrections. It also sets up ``git`` for it, and it
|
||||
brings the fork up to date first:
|
||||
|
||||
* The manifest records the email of the uploader. ``czdev`` takes the public email of the
|
||||
GitHub profile. MeshTerm uses only ``johnputer@meshterm.net`` (JP, 2026-10-07), so the
|
||||
@@ -17,6 +18,13 @@ pull request. The store merges an update by itself when its checks pass. This sc
|
||||
* ``czdev`` pushes to ``git@github.com:``. A CI runner or a Cardputer has no SSH key for
|
||||
GitHub, so the script sends ``git`` over HTTPS with the token, and gives it a name and an
|
||||
email for the commit. These settings are in the environment of this process only.
|
||||
* ``czdev`` starts its branch from the default branch of the fork, and the store's check
|
||||
compares the pull request's branch with the store's ``main`` directly. Thus, if the fork
|
||||
is behind, each file that the store removed since then looks like a new file of the pull
|
||||
request, and the check fails on it. This occurred on 2026-10-12, when M5Stack removed
|
||||
all apps from the store to start the reviews again. The script brings the fork's
|
||||
default branch up to the store's ``main`` before ``czdev`` starts. If it cannot, it
|
||||
publishes nothing.
|
||||
|
||||
The corrections replace two functions of ``czdev``. Thus the release workflow pins
|
||||
AppBuilder to one commit. If a later ``czdev`` does not have these functions, the script
|
||||
@@ -43,6 +51,7 @@ import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
@@ -56,10 +65,11 @@ PUBLISHED_INDEX = "https://cardputerzero.github.io/packages/dists/stable/main/bi
|
||||
TOKEN_ENV = "CARDPUTER_STORE_TOKEN"
|
||||
|
||||
|
||||
def _api(path: str, token: str) -> object:
|
||||
"""GET one GitHub API path, as JSON."""
|
||||
def _api(path: str, token: str, body: dict[str, str] | None = None) -> object:
|
||||
"""Call one GitHub API path, as JSON: a GET, or a POST of ``body``."""
|
||||
request = urllib.request.Request(
|
||||
f"https://api.github.com{path}",
|
||||
data=None if body is None else json.dumps(body).encode(),
|
||||
headers={"Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json"},
|
||||
)
|
||||
with urllib.request.urlopen(request, timeout=30) as response: # noqa: S310 - a fixed https host
|
||||
@@ -98,6 +108,30 @@ def _open_pull_requests(package: str, login: str, token: str) -> dict[str, str]:
|
||||
}
|
||||
|
||||
|
||||
def _sync_fork(login: str, token: str) -> str | None:
|
||||
"""Bring the fork's default branch up to the store's ``main``. Return a problem, or ``None``.
|
||||
|
||||
GitHub's merge-upstream does a fast-forward when the fork has no commits of its own. If
|
||||
it has some and the merge fails, the script must not publish (refer to the module
|
||||
docstring). With no fork, there is nothing to do: ``czdev`` makes one from the store's
|
||||
current ``main``.
|
||||
"""
|
||||
try:
|
||||
fork = _api(f"/repos/{login}/{TARGET.split('/')[1]}", token)
|
||||
except urllib.error.HTTPError as error:
|
||||
if error.code == 404:
|
||||
return None
|
||||
raise
|
||||
if (fork.get("parent") or {}).get("full_name") != TARGET:
|
||||
return f"{fork['full_name']} is not a fork of {TARGET}"
|
||||
branch = fork["default_branch"]
|
||||
try:
|
||||
_api(f"/repos/{fork['full_name']}/merge-upstream", token, {"branch": branch})
|
||||
except urllib.error.HTTPError as error:
|
||||
return f"cannot bring {fork['full_name']} {branch} up to {TARGET} (HTTP {error.code})"
|
||||
return None
|
||||
|
||||
|
||||
def _set_up_git(login: str, token: str) -> None:
|
||||
"""Send ``czdev``'s git over HTTPS with ``token``, as MeshTerm's author."""
|
||||
os.environ["CZDEV_GIT_TOKEN"] = token
|
||||
@@ -181,6 +215,10 @@ def main() -> int:
|
||||
for other, url in pending.items():
|
||||
print(f"{mark}{other} is still open ({url}); {version} waits for its merge")
|
||||
return 0
|
||||
problem = _sync_fork(login, token)
|
||||
if problem:
|
||||
sys.stderr.write(f"store-publish: {problem}\n")
|
||||
return 1
|
||||
|
||||
_correct(publish, GitHubClient)
|
||||
_set_up_git(login, token)
|
||||
|
||||
@@ -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
@@ -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(
|
||||
|
||||
@@ -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
|
||||
Reference in new issue
Block a user