Author SHA1 Message Date
richandClaude Opus 5.5 d9426166c8 Search every emoji by name in the reaction dialog
CI / lint (push) Canceled after 0s
CI / test macos-latest py3.14 (push) Canceled after 0s
CI / test ubuntu-24.04 py3.10 (push) Canceled after 0s
CI / test ubuntu-latest py3.11 (push) Canceled after 0s
CI / test ubuntu-latest py3.12 (push) Canceled after 0s
CI / test ubuntu-latest py3.13 (push) Canceled after 0s
CI / test ubuntu-latest py3.14 (push) Canceled after 0s
CI / test windows-latest py3.10 (push) Canceled after 0s
CI / test windows-latest py3.14 (push) Canceled after 0s
CI / build (push) Canceled after 0s
Issue trailers / check (push) Canceled after 0s
^E still lists the reaction_emoji preference first, under "Your reactions". Under
"All emoji" it now lists each other emoji with its Unicode name, and the list
filters as you type: "fire" finds 🔥, "heart" finds the hearts. The catalog comes
from the Unicode database of the Python that runs MeshTerm and from Rich's table of
text-default emoji, so it needs no data file.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 07:47:13 -04:00
richandClaude Opus 5.5 f3aa8e5aef Allow channel sender names of up to 31 characters
The firmware keeps a node name in 32 bytes, but the sender prefix and the @[Name]
mention matched only 20 characters. A longer name stayed in the body: the transcript
showed no sender for it, and its reactions showed as new messages. Found on hardware
with "Hazelwood r0b3.Anduril" (22 characters).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 07:47:13 -04:00
richandClaude Haiku 5.5 a323342971 Add emoji reactions in the wire format of MeshCore One
Select a message in a channel or a DM and press ^E (Shift+F2 on the handhelds) to
react with an emoji. A reaction is a usual message on the wire: MeshTerm writes the
format of MeshCore One and reads it and the two formats of meshcore-open. The
transcript folds each reaction under the message that it names, with a count, or
the names of the nodes while that message is selected. A reaction does not raise
the unread badge, and the Messages list previews it as "reacted 👍".

A reaction names its target by the sender's timestamp. Thus MeshTerm now gives the
device the timestamp of each message it sends and stores exactly that second. A
retry of a DM keeps the timestamp and counts the try number, as the protocol
specifies. The emoji that ^E offers are the reaction_emoji preference. Rooms have
no reactions.

Co-Authored-By: Claude Haiku 5.5 <noreply@anthropic.com>
2026-10-15 07:47:13 -04:00
Jean-Pierre MartineauandClaude Opus 5.5 488db7f278 MeshTerm 0.11.1
Installers / cardputer-zero-deb (push) Canceled after 0s
Installers / windows-x64 (push) Canceled after 0s
Installers / macos-x64 (push) Canceled after 0s
Installers / macos-arm64 (push) Canceled after 0s
Installers / linux-x64 (push) Canceled after 0s
Installers / linux-arm64 (push) Canceled after 0s
GitHub release / build (push) Canceled after 0s
GitHub release / create the release (push) Canceled after 0s
GitHub release / send the Cardputer package to the app store (push) Canceled after 0s
GitHub release / put the version on meshterm.net (push) Canceled after 0s
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-10-13 00:06:30 -04:00
Jean-Pierre MartineauandClaude Opus 5.5 b434cb1643 The changelog names #26 under the identity key fixes
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-10-12 23:53:36 -04:00
Jean-Pierre MartineauandClaude Haiku 5.5 723f7bdfef 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>
2026-10-12 23:51:03 -04:00
Jean-Pierre MartineauandClaude Opus 5.5 7209d89911 The store job brings JP's fork of the store up to date before it publishes
czdev starts its branch from the fork's default branch, and the store's
check compares the PR branch with the store's main directly. After
M5Stack removed every app on 2026-10-12, the fork was 10 commits behind,
so the 0.11.0 PR (#176) carried all 40 removed apps as new files, and
the check failed on the first of them. store-publish.py now calls
GitHub's merge-upstream for the fork first, and publishes nothing if
that fails. PR #176 was repaired by hand with "Update branch".

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-10-12 10:57:20 -04:00
32 changed files with 2368 additions and 114 deletions

No files matched your search

+4 -1
View File
@@ -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
View File
@@ -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
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. |
@@ -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 |
+1
View File
@@ -194,6 +194,7 @@ and in quoted text.
| packet | One radio transmission | frame (for radio data) |
| protocol frame | One unit of the companion protocol on USB, Bluetooth, or TCP: a command, a reply, or a push from the companion. In code that handles only this protocol, "frame" alone is permitted | packet (for this) |
| message | A text message, direct or on a channel | |
| reaction | An emoji that a node sends to answer one message. It is a message on the wire, and the chat shows it under the message that it names | |
| send | MeshTerm gives a message, a packet, or a command to the device | emit |
| transmit | A radio puts a packet on the air | |
| broadcast | Transmit to all nodes, with no destination | |
+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
+1 -1
View File
@@ -15,7 +15,7 @@ This page lists each interactive screen with its scripted equivalent, where one
| Feature | What it does | Scripted |
| --- | --- | --- |
| **💬 Chat** | Live full-screen messages on channels and to contacts. The transcript scrolls, the input line stays at the bottom, and sent and received messages appear together. The rooms that you joined are also here. Each post shows under the name of its author. MeshTerm stores each message. Unread counts show in the menu header. | `meshterm chat send / history / list / listen` |
| **💬 Chat** | Live full-screen messages on channels and to contacts. The transcript scrolls, the input line stays at the bottom, and sent and received messages appear together. The rooms that you joined are also here. Each post shows under the name of its author. Select a message and press **^E** to react to it with an emoji. The reaction shows under the message, and MeshCore One and meshcore-open show it too. The emoji that ^E offers are a preference. MeshTerm stores each message. Unread counts show in the menu header. | `meshterm chat send / history / list / listen` |
| **📻 Channels** | Create, join, reorder, mute, and share mesh channels. You can share a channel with a QR code or a `meshcore://` link. If you give a channel a **send scope**, its messages are flooded into one region. Only the repeaters that carry that region relay them. When you change to a new device, export all the channels to a file, then import the file on the new device. The **channel guide** lists the public channels that MeshTerm heard, also the channels that are not on your device, and adds one with Enter. | `meshterm channels list / add / join / import / share / clear / scope / export / guide` |
| **📌 Rooms** | Each room server that your radio knows. A room server is a message board on a radio. It keeps the latest posts and sends you the posts that you missed. Join a room with its password. If a room does not answer, MeshTerm tells you what it knows and offers the next step to try. For example, it can send again by flood when the route that your radio learned is no longer good. | `meshterm rooms list / join / forget` |
| **📨 Courier** | A store-and-forward outbox for contacts that you cannot reach yet. Queue a message. MeshTerm sends it when the contact is next heard, or at a time that you set. It tracks the acks, and it waits longer between each try (exponential backoff). | `meshterm courier queue / list / send / cancel / clear` |
+1 -1
View File
@@ -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.
+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)
+9 -5
View File
@@ -35,17 +35,21 @@ from urllib.parse import parse_qs, quote, urlsplit
# not one time for each candidate channel. Thus the reception path pays almost nothing.
#: Matches the ``Name: message`` convention that channel senders use to identify themselves
#: (the protocol has no sender field). The name is 1–20 characters that are not colons, and
#: (the protocol has no sender field). The name is 1–31 characters that are not colons, and
#: ``": "`` must follow it. This rule is strict enough that it does not match ``http://…``
#: or ``note:x``.
SENDER_PREFIX = re.compile(r"^([^\s:][^:]{0,19}):[ \t]+(.*)$", re.DOTALL)
#: or ``note:x``. The firmware keeps a node name in 32 bytes with its terminator, so 31 is
#: the longest name that a sender can have. A limit of 20 left
#: a longer name in the body. Then the transcript showed no sender for it, and a reaction
#: from it did not read as a reaction.
SENDER_PREFIX = re.compile(r"^([^\s:][^:]{0,30}):[ \t]+(.*)$", re.DOTALL)
#: Matches an ``@[Name]`` mention token, which the reply flow puts into the compose line in
#: advance. A transcript renders each token as a bare ``@Name`` in the hue of that sender,
#: and does not show the brackets. The history face of the CLI also takes out the same
#: names. Thus the token is defined here, with the remainder of the message-body parsing, and
#: not in one of the two renderers. The name is 1–20 characters that are not ``]``.
MENTION = re.compile(r"@\[([^\]]{1,20})\]")
#: not in one of the two renderers. The name is 1–31 characters that are not ``]``, the
#: same limit as :data:`SENDER_PREFIX`.
MENTION = re.compile(r"@\[([^\]]{1,31})\]")
def split_channel_sender(text: str) -> tuple[str | None, str]:
+57 -15
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."""
@@ -1551,12 +1557,21 @@ class Device(ABC):
# -- messaging ---------------------------------------------------------------
@abstractmethod
async def send_direct_message(self, contact: Contact, text: str) -> Delivery:
async def send_direct_message(
self, contact: Contact, text: str, *, timestamp: int | None = None, attempt: int = 0
) -> Delivery:
"""Send a direct text message to a contact, and wait some time for its ack.
Args:
contact: The recipient. Its ``public_key`` addresses the message.
text: The message body.
timestamp: The timestamp that the message carries, in seconds. ``None`` uses
the clock of this computer. The recipient sees this value, and a reaction
names the message by it (refer to :mod:`~meshterm.core.reactions`). Thus the
caller gives it when it must know the value.
attempt: The number of the try (0 to 3). A retry of the same message has the
same timestamp and the next number, as the protocol specifies. The number
makes the packet of the retry different from the first one.
Returns:
The :class:`~meshterm.core.models.Delivery`: the ack code that the radio
@@ -1568,12 +1583,16 @@ class Device(ABC):
"""
@abstractmethod
async def send_channel_message(self, index: int, text: str) -> None:
async def send_channel_message(
self, index: int, text: str, *, timestamp: int | None = None
) -> None:
"""Broadcast a text message on a channel slot.
Args:
index: The zero-based channel slot on which to transmit.
text: The message body.
timestamp: The timestamp that the message carries, in seconds. ``None`` uses
the clock of this computer (refer to :meth:`send_direct_message`).
Raises:
DeviceCommandError: If the companion rejected the send.
@@ -1625,7 +1644,9 @@ class Device(ABC):
self._scope_leaked = None
yield
async def send_channel_in_scope(self, index: int, text: str, scope: str | None) -> None:
async def send_channel_in_scope(
self, index: int, text: str, scope: str | None, *, timestamp: int | None = None
) -> None:
"""Broadcast on a channel in a given scope, and leave the session scope as it was.
This is the only way that MeshTerm sends a channel message. ``None`` is the plain
@@ -1653,6 +1674,8 @@ class Device(ABC):
text: The message body.
scope: A region name, :data:`~meshterm.core.regions.WILDCARD` for unscoped, or
``None`` for the default of the device.
timestamp: The timestamp that the message carries (refer to
:meth:`send_channel_message`).
Raises:
FloodScopeError: If the companion cannot send in ``scope``. Nothing was sent.
@@ -1665,10 +1688,10 @@ class Device(ABC):
bare = validate_region(bare)
async with self.transmitting():
if not bare:
await self.send_channel_message(index, text)
await self.send_channel_message(index, text, timestamp=timestamp)
return
if bare == REGION_WILDCARD and await self._unscoped_is_plain():
await self.send_channel_message(index, text)
await self.send_channel_message(index, text, timestamp=timestamp)
return
try:
await self.set_flood_scope(bare)
@@ -1676,7 +1699,7 @@ class Device(ABC):
raise FloodScopeError(bare, _scope_refusal(bare, exc)) from exc
self._scope_active = True
try:
await self.send_channel_message(index, text)
await self.send_channel_message(index, text, timestamp=timestamp)
finally:
self._scope_active = False
try:
@@ -1995,7 +2018,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:
@@ -4106,7 +4134,7 @@ class MeshCoreDevice(Device):
return stop_pump
async def send_direct_message( # noqa: D102 - inherited docstring
self, contact: Contact, text: str
self, contact: Contact, text: str, *, timestamp: int | None = None, attempt: int = 0
) -> Delivery:
transmit_gate.mark()
from meshcore import EventType
@@ -4135,7 +4163,9 @@ class MeshCoreDevice(Device):
try:
# Only the hand-over holds the transmit lock. The ack wait below is not a send.
async with self.transmitting():
result = await mc.commands.send_msg(pub, text)
result = await mc.commands.send_msg(
pub, text, timestamp=timestamp, attempt=attempt & 3
)
if result is None or getattr(result, "is_error", lambda: False)():
# A recipient for which the firmware has no entry is the only rejection with
# an obvious repair. Thus it gets its own class, so the chat screen can offer
@@ -4191,11 +4221,11 @@ class MeshCoreDevice(Device):
return Delivery(code[0], ack, flood)
async def send_channel_message( # noqa: D102 - inherited docstring
self, index: int, text: str
self, index: int, text: str, *, timestamp: int | None = None
) -> None:
transmit_gate.mark()
async with self.transmitting():
self._ok(await self._require().commands.send_chan_msg(index, text))
self._ok(await self._require().commands.send_chan_msg(index, text, timestamp))
@staticmethod
def _ok(event): # type: ignore[no-untyped-def]
@@ -4555,7 +4585,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()
@@ -4863,7 +4905,7 @@ class MockDevice(Device):
self._tx_power = value
async def send_direct_message( # noqa: D102 - inherited docstring
self, contact: Contact, text: str
self, contact: Contact, text: str, *, timestamp: int | None = None, attempt: int = 0
) -> Ack | None:
transmit_gate.mark()
async with self.transmitting():
@@ -4908,7 +4950,7 @@ class MockDevice(Device):
task.add_done_callback(self._bg_tasks.discard)
async def send_channel_message( # noqa: D102 - inherited docstring
self, index: int, text: str
self, index: int, text: str, *, timestamp: int | None = None
) -> None:
transmit_gate.mark()
async with self.transmitting():
+11
View File
@@ -55,6 +55,7 @@ from .atomicwrite import write_atomically
from .courier_store import DONE_CAP
from .geo import DEFAULT_VIEW_FRACTION
from .models import NAME_OVERLAP_DAYS
from .reactions import DEFAULT_REACTIONS
from .watch_store import ALERT_CAP, DEFAULT_SILENCE_HOURS, OFF, SILENCE_CHOICES_H
if sys.version_info >= (3, 11):
@@ -290,6 +291,16 @@ PREFERENCES: tuple[PrefSpec, ...] = (
minimum=0,
maximum=2,
),
PrefSpec(
key="reaction_emoji",
label="Reactions",
help="Emoji to offer when you react to a message",
group="Sending",
value_type="str",
# The default set of MeshCore One. A word that is not one emoji is ignored when the
# chat reads the list (refer to meshterm.core.reactions.reaction_set).
default=DEFAULT_REACTIONS,
),
PrefSpec(
key="room_relogin_minutes",
label="Room re-login",
+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."
)
+538
View File
@@ -0,0 +1,538 @@
# SPDX-License-Identifier: Apache-2.0
"""Emoji reactions to chat messages, in the wire format of MeshCore One.
A reaction is a usual text message. The firmware and the repeaters do not know the format.
Only the apps do. MeshTerm writes the format of MeshCore One, and reads it and the two
formats of meshcore-open. The protocol is in ``docs/Reactions.md`` of MeshCore One. In short:
* A reaction names its target message by a hash of the body of the target and the
timestamp of its sender: SHA-256 of the UTF-8 body and the timestamp as a little-endian
32-bit integer, the first five bytes, in Crockford Base32 (eight characters). Each node
that received the target calculates the same hash, because the sender's timestamp travels
with the message.
* On a channel, the reaction also names the author of the target, because two nodes can
send the same body in the same second: ``@[Author]👍`` then a newline and the hash.
Before MeshCore One 1.4.1, the emoji came first (``👍@[Author]``). MeshTerm reads the two
orders and writes the new one.
* In a direct exchange, there are two persons only, so the reaction is ``👍``, a newline,
and the hash.
A body that only starts with an emoji is a usual message. The emoji part must be exactly
one emoji (one glyph) for the body to be a reaction.
MeshTerm stores a reaction as the usual message that it is. Thus the history keeps all that
was received. The chat screen folds the reactions of a transcript onto their targets when it
shows the transcript (refer to :class:`ReactionFold`).
"""
from __future__ import annotations
import hashlib
import unicodedata
from dataclasses import dataclass, field
from datetime import datetime
from .channels import split_channel_sender
from .models import ChatMessage
#: The reactions that the reaction dialog offers when the preference is empty. They are the
#: default set of MeshCore One, in its order.
DEFAULT_REACTIONS = "👍 👎 ❤️ 😂 😮 😢"
#: How far apart in time, in seconds, a reaction of meshcore-open and its target can be. The
#: short hash of meshcore-open (16 bits) matches many messages in a long history. Thus
#: MeshTerm compares it only with the messages near the reaction in time. MeshCore One uses
#: the same window.
WINDOW_S = 300
#: The maximum number of reactions that wait for a target that did not arrive yet.
PENDING_CAP = 100
# -- Crockford Base32 ---------------------------------------------------------------------
_CROCKFORD = "0123456789abcdefghjkmnpqrstvwxyz"
_DECODE = {char: char for char in _CROCKFORD}
_DECODE.update({char.upper(): char for char in _CROCKFORD})
# The letters that a person can mistake for a digit read as that digit.
_DECODE.update({"o": "0", "O": "0", "i": "1", "I": "1", "l": "1", "L": "1"})
def _crockford(data: bytes) -> str:
"""Encode five bytes as eight Crockford Base32 characters, the most significant first."""
bits = int.from_bytes(data[:5], "big")
return "".join(_CROCKFORD[(bits >> shift) & 0x1F] for shift in range(35, -5, -5))
def _normal_hash(raw: str) -> str | None:
"""The canonical (lowercase) form of an eight-character hash, or ``None`` if it is not one."""
if len(raw) != 8:
return None
out = [_DECODE.get(char) for char in raw]
if None in out:
return None
return "".join(out) # type: ignore[arg-type] - None is excluded above
def message_hash(body: str, timestamp: int) -> str:
"""The hash with which a reaction names its target message.
Args:
body: The body of the target. On a channel, this is the text after the ``Name: ``
prefix of the sender.
timestamp: The timestamp that the sender gave the target, in seconds.
Returns:
Eight lowercase Crockford Base32 characters.
"""
data = body.encode("utf-8") + (timestamp & 0xFFFFFFFF).to_bytes(4, "little")
return _crockford(hashlib.sha256(data).digest())
def wire_timestamp(when: datetime) -> int:
"""The timestamp of a message as its sender put it on the wire: whole seconds."""
return int(when.timestamp())
# -- the emoji part -----------------------------------------------------------------------
_VS = frozenset("︎️")
_ZWJ = "‍"
_KEYCAP = "⃣"
_TONES = frozenset(chr(cp) for cp in range(0x1F3FB, 0x1F400))
_TAGS = frozenset(chr(cp) for cp in range(0xE0020, 0xE0080))
def _is_regional(char: str) -> bool:
"""Whether ``char`` is a Regional Indicator (one half of a flag)."""
return 0x1F1E6 <= ord(char) <= 0x1F1FF
def _is_pictograph(char: str) -> bool:
"""Whether ``char`` can start an emoji: a symbol outside the Latin and punctuation blocks."""
return ord(char) >= 0x2000 and unicodedata.category(char) == "So"
def is_reaction_emoji(value: str) -> bool:
"""Whether ``value`` is exactly one emoji: the emoji part of a reaction.
One emoji is one glyph. It can be a flag (two Regional Indicators), a keycap (``1️⃣``),
or pictographs joined with zero-width joiners (``❤️‍🔥``). Each pictograph can have a
variation selector, a skin tone, and a tag sequence. Text, or an emoji with text after
it, is not a reaction.
"""
if not value or value.isascii():
return False
if len(value) == 2 and _is_regional(value[0]) and _is_regional(value[1]):
return True
if value[0] in "0123456789#*":
return value[1:] in (_KEYCAP, "️" + _KEYCAP)
for part in value.split(_ZWJ):
if not part or not _is_pictograph(part[0]):
return False
if any(char not in _VS and char not in _TONES and char not in _TAGS for char in part[1:]):
return False
return True
def emoji_name(emoji: str) -> str:
"""A name for an emoji, for a person to read: its Unicode name in lowercase."""
try:
return unicodedata.name(emoji[0]).lower()
except (ValueError, IndexError):
return "emoji"
def reaction_set(preference: str) -> list[str]:
"""The emoji that the reaction dialog offers, from the ``reaction_emoji`` preference.
The preference is a list of emoji with spaces between them. A word that is not one emoji
is ignored, and so is a second copy of an emoji. An empty result gives
:data:`DEFAULT_REACTIONS`, because a reaction dialog with no rows is of no use.
"""
seen: list[str] = []
for word in (preference or "").split():
if is_reaction_emoji(word) and word not in seen:
seen.append(word)
return seen or DEFAULT_REACTIONS.split()
# -- the wire format ----------------------------------------------------------------------
#: The three forms of a reaction on the wire.
MC1 = "mc1" # MeshCore One: an eight-character hash
MCO = "mco" # meshcore-open v3: a 16-bit hash, and an index into its table of emoji
MCO_V1 = "mco1" # meshcore-open v1 (before January 2026): the time and two 30-bit hashes
@dataclass(frozen=True, slots=True)
class Reaction:
"""A reaction, as read from the body of a message.
Attributes:
emoji: The emoji of the reaction.
form: :data:`MC1`, :data:`MCO`, or :data:`MCO_V1`.
hash: How the reaction names its target. For :data:`MC1`, the eight-character hash
(refer to :func:`message_hash`). For :data:`MCO`, four lowercase hex digits.
For :data:`MCO_V1`, the message id of meshcore-open.
author: For a channel reaction of :data:`MC1`, the name of the author of the
target. ``None`` in all other cases.
timestamp: For :data:`MCO_V1`, the timestamp of the target, in seconds.
author_hash: For :data:`MCO_V1`, the Dart hash of the name of the author.
body_hash: For :data:`MCO_V1`, the Dart hash of the body of the target.
"""
emoji: str
form: str
hash: str
author: str | None = None
timestamp: int | None = None
author_hash: int | None = None
body_hash: int | None = None
def parse(body: str, *, channel: bool) -> Reaction | None:
"""Read a message body as a reaction.
Args:
body: The body of the message. On a channel, this is the text after the
``Name: `` prefix of the sender.
channel: Whether the message is on a channel. A channel reaction of MeshCore One
names the author of its target, and a direct one does not.
Returns:
The reaction, or ``None`` when the body is a usual message.
"""
if body.startswith("r:"):
return _parse_mco(body) or _parse_mco_v1(body)
head, newline, tail = body.rpartition("\n")
if not newline:
return None
digest = _normal_hash(tail)
if digest is None:
return None
if not channel:
if "@[" in head or not is_reaction_emoji(head):
return None
return Reaction(head, MC1, digest)
before, mark, after = head.partition("@[")
if not mark:
return None
if before:
# The order before MeshCore One 1.4.1: the emoji, then the mention.
if not after.endswith("]"):
return None
emoji, author = before, after[:-1]
else:
author, close, emoji = after.partition("]")
if not close:
return None
if not author or not is_reaction_emoji(emoji):
return None
return Reaction(emoji, MC1, digest, author=author)
def channel_text(emoji: str, author: str, body: str, timestamp: int) -> str:
"""The body of a channel reaction to the message ``body`` of ``author``.
Args:
emoji: The emoji of the reaction.
author: The name of the author of the target, as the target carried it.
body: The body of the target, without the ``Name: `` prefix.
timestamp: The timestamp that the author gave the target.
"""
return f"@[{author}]{emoji}\n{message_hash(body, timestamp)}"
def direct_text(emoji: str, body: str, timestamp: int) -> str:
"""The body of a direct reaction to the message ``body`` sent at ``timestamp``."""
return f"{emoji}\n{message_hash(body, timestamp)}"
# -- meshcore-open ------------------------------------------------------------------------
#: The table of emoji of meshcore-open (its ``emoji_picker.dart``), in order: the quick set,
#: then the smileys, gestures, hearts, and objects. A reaction of v3 sends an index into it.
# fmt: off
MCO_EMOJI: tuple[str, ...] = (
"👍", "❤️", "😂", "🎉", "👏", "🔥",
"😀", "😃", "😄", "😁", "😅", "😂", "🤣", "😊",
"😇", "🙂", "🙃", "😉", "😌", "😍", "🥰", "😘",
"😗", "😙", "😚", "😋", "😛", "😝", "😜", "🤪",
"🤨", "🧐", "🤓", "😎", "🥸", "🤩", "🥳", "😏",
"😒", "😞", "😔", "😟", "😕", "🙁", "😣", "😖",
"😫", "😩", "🥺", "😢", "😭", "😤", "😠", "😡",
"🤬", "🤯", "😳", "🥵", "🥶", "😱", "😨", "😰",
"😥", "😓", "🤗", "🤔", "🤭", "🤫", "🤥", "😶",
"👍", "👎", "👊", "✊", "🤛", "🤜", "🤞", "✌️",
"🤟", "🤘", "👌", "🤌", "🤏", "👈", "👉", "👆",
"👇", "☝️", "👋", "🤚", "🖐️", "✋", "🖖", "👏",
"🙌", "👐", "🤲", "🤝", "🙏", "✍️", "💅", "🤳",
"💪",
"❤️", "🧡", "💛", "💚", "💙", "💜", "🖤", "🤍",
"🤎", "💔", "❤️‍🔥", "❤️‍🩹", "💕", "💞", "💓", "💗",
"💖", "💘", "💝", "💟", "💌", "💢", "💥", "💫",
"💦", "💨", "🕳️", "💬", "👁️‍🗨️", "🗨️", "🗯️", "💭",
"🎉", "🎊", "🎈", "🎁", "🎀", "🪅", "🪆", "🏆",
"🥇", "🥈", "🥉", "⚽", "⚾", "🥎", "🏀", "🏐",
"🏈", "🏉", "🎾", "🥏", "🎳", "🏏", "🏑", "🏒",
"🥍", "🏓", "🏸", "🥊", "🥋", "🥅", "⛳", "🔥",
"⭐", "🌟", "✨", "⚡", "💡", "🔦", "🏮", "🪔",
"📱", "💻", "⌚", "📷", "📺", "📻", "🎵", "🎶",
"🚀",
)
# fmt: on
_HEX = frozenset("0123456789abcdef")
def _parse_mco(body: str) -> Reaction | None:
"""Read ``r:{hash}:{index}``, a reaction of meshcore-open v3 (four and two hex digits)."""
if len(body) != 9 or body[6] != ":":
return None
digest, index = body[2:6], body[7:]
if not set(digest + index) <= _HEX:
return None
position = int(index, 16)
if position >= len(MCO_EMOJI):
return None
return Reaction(MCO_EMOJI[position], MCO, digest)
def _parse_mco_v1(body: str) -> Reaction | None:
"""Read ``r:{millis}_{author hash}_{body hash}:{emoji}``, a reaction of meshcore-open v1."""
ident, colon, emoji = body[2:].rpartition(":")
if not colon or not ident or not emoji:
return None
parts = ident.split("_")
if len(parts) != 3 or not all(part.isdigit() for part in parts):
return None
seconds = int(parts[0]) // 1000
if seconds > 0xFFFFFFFF:
return None
return Reaction(
emoji,
MCO_V1,
ident,
timestamp=seconds,
author_hash=int(parts[1]),
body_hash=int(parts[2]),
)
def dart_hash(units: list[int]) -> int:
"""The ``String.hashCode`` of the Dart VM over UTF-16 code units: a 30-bit value, never 0."""
mask = 0xFFFFFFFF
value = 0
for unit in units:
value = (value + unit) & mask
value = (value + (value << 10)) & mask
value ^= value >> 6
value = (value + (value << 3)) & mask
value ^= value >> 11
value = (value + (value << 15)) & mask
value &= (1 << 30) - 1
return value or 1
def _units(text: str) -> list[int]:
"""The UTF-16 code units of ``text``."""
raw = text.encode("utf-16-le")
return [int.from_bytes(raw[i : i + 2], "little") for i in range(0, len(raw), 2)]
def mco_hash(timestamp: int, author: str | None, body: str) -> str:
"""The 16-bit hash with which meshcore-open v3 names a target, as four hex digits.
It is the Dart hash of the timestamp in decimal, the name of the author (on a channel
only), and the first five UTF-16 code units of the body.
"""
units = _units(str(timestamp)) + (_units(author) if author else []) + _units(body)[:5]
return f"{dart_hash(units) & 0xFFFF:04x}"
# -- the fold -----------------------------------------------------------------------------
@dataclass(frozen=True, slots=True)
class Reactor:
"""Who sent a reaction: our node, or another node by the name that it gave.
Attributes:
name: The name of the node. ``None`` for our node, and for the peer of a direct
exchange (the conversation names the peer).
ours: Whether our node sent the reaction.
"""
name: str | None
ours: bool = False
@dataclass(slots=True)
class Badge:
"""One emoji on one message, and the nodes that reacted with it, the first first."""
emoji: str
reactors: list[Reactor] = field(default_factory=list)
@property
def ours(self) -> bool:
"""Whether our node is one of the nodes that reacted with this emoji."""
return any(reactor.ours for reactor in self.reactors)
@dataclass(slots=True)
class _Target:
"""A message of the transcript, as the fold matches reactions against it."""
message: ChatMessage
author: str | None
body: str
timestamp: int
class ReactionFold:
"""Fold the reactions of one conversation onto the messages that they react to.
Give each message of the transcript to :meth:`take`, in the order of arrival. A message
that is not a reaction becomes a target, and :meth:`take` returns ``False``. A reaction
attaches to its target, and :meth:`take` returns ``True``. Then the transcript does not
show it as a message. A reaction whose target is not known yet waits (at most
:data:`PENDING_CAP` of them), because on a mesh a reaction can arrive before its target.
:meth:`badges` gives the reactions on one message.
The fold counts each emoji one time for each node. A node can send the same reaction
again, and a repeater can deliver a copy two times.
"""
def __init__(self, *, channel: bool, self_name: str | None = None) -> None:
"""Make an empty fold.
Args:
channel: Whether the conversation is a channel. A channel reaction names the
author of its target, and a direct one does not.
self_name: The name of our node. On a channel, it is the author of each message
that we sent. Without it, the reactions to our channel messages of
MeshCore One cannot find their target.
"""
self._channel = channel
self._self_name = self_name
self._targets: list[_Target] = []
self._by_id: dict[int, _Target] = {}
self._by_hash: dict[tuple[str | None, str], _Target] = {}
self._badges: dict[int, list[Badge]] = {}
self._pending: list[tuple[Reaction, Reactor]] = []
def take(self, message: ChatMessage) -> bool:
"""Fold one message in, and return whether it is a reaction.
Args:
message: The next message of the transcript.
Returns:
``True`` when the message is a reaction. The transcript then does not show it.
``False`` when it is a usual message, which other reactions can now name.
"""
author, body = self._author_and_body(message)
reaction = parse(body, channel=self._channel)
if reaction is None:
self._index(message, author, body)
return False
reactor = Reactor(
None if message.outbound else (author if self._channel else None), ours=message.outbound
)
target = self._find(reaction, wire_timestamp(message.created_at))
if target is not None:
self._attach(target.message, reaction.emoji, reactor)
elif reaction.form == MC1:
self._pending.append((reaction, reactor))
del self._pending[:-PENDING_CAP]
return True
def badges(self, message: ChatMessage) -> list[Badge]:
"""The reactions on ``message``: the most frequent emoji first, then the earliest."""
badges = self._badges.get(id(message), [])
order = {id(badge): index for index, badge in enumerate(badges)}
return sorted(badges, key=lambda badge: (-len(badge.reactors), order[id(badge)]))
def count(self, message: ChatMessage) -> int:
"""The number of reactions on ``message``. A render cache compares it to see a change."""
return sum(len(badge.reactors) for badge in self._badges.get(id(message), []))
def reacted(self, message: ChatMessage, emoji: str) -> bool:
"""Whether our node already reacted to ``message`` with ``emoji``."""
return any(
badge.emoji == emoji and badge.ours for badge in self._badges.get(id(message), [])
)
def target_of(self, message: ChatMessage) -> tuple[str | None, str, int] | None:
"""The author, the body, and the timestamp with which a reaction names ``message``.
``None`` when ``message`` is not a target of this fold (a reaction, or a message
that the fold never took).
"""
target = self._by_id.get(id(message))
if target is None or target.message is not message:
return None
return target.author, target.body, target.timestamp
def _author_and_body(self, message: ChatMessage) -> tuple[str | None, str]:
"""The author and the body of a message, as the hash of a reaction sees them.
On a channel, the body of a received message is the text after the ``Name: ``
prefix, without the spaces at its two ends. Our channel messages have no prefix in
the history, and their author is our node.
"""
if not self._channel:
return None, message.text
if message.outbound:
return self._self_name, message.text.strip(" \t")
name, body = split_channel_sender(message.text)
return name, body.strip(" \t")
def _index(self, message: ChatMessage, author: str | None, body: str) -> None:
"""Make ``message`` a target, and attach the reactions that waited for it."""
target = _Target(message, author, body, wire_timestamp(message.created_at))
self._targets.append(target)
self._by_id[id(message)] = target
if self._channel and not author:
return # a channel reaction of MeshCore One must name the author
key = (author if self._channel else None, message_hash(body, target.timestamp))
self._by_hash[key] = target
waiting = [entry for entry in self._pending if (entry[0].author, entry[0].hash) == key]
if waiting:
self._pending = [entry for entry in self._pending if entry not in waiting]
for reaction, reactor in waiting:
self._attach(message, reaction.emoji, reactor)
def _find(self, reaction: Reaction, at: int) -> _Target | None:
"""The target that ``reaction`` names, or ``None``. The newest target wins a tie."""
if reaction.form == MC1:
return self._by_hash.get((reaction.author, reaction.hash))
if reaction.form == MCO:
anchor = at
else:
anchor = reaction.timestamp or 0
for target in reversed(self._targets):
if abs(target.timestamp - anchor) > WINDOW_S:
continue
if reaction.form == MCO:
author = target.author if self._channel else None
if mco_hash(target.timestamp, author, target.body) == reaction.hash:
return target
elif dart_hash(_units(target.body)) == reaction.body_hash:
if self._channel and target.author is not None:
if dart_hash(_units(target.author)) != reaction.author_hash:
continue
return target
return None
def _attach(self, message: ChatMessage, emoji: str, reactor: Reactor) -> None:
"""Add one reaction to ``message``, if that node did not already send that emoji."""
badges = self._badges.setdefault(id(message), [])
badge = next((b for b in badges if b.emoji == emoji), None)
if badge is None:
badge = Badge(emoji)
badges.append(badge)
if reactor not in badge.reactors:
badge.reactors.append(reactor)
+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"]))
+70 -12
View File
@@ -19,11 +19,14 @@ MeshTerm always stores them while a device listens.
from __future__ import annotations
import asyncio
import time
from collections import OrderedDict
from collections.abc import Callable
from datetime import datetime, timezone
from typing import TYPE_CHECKING
from ..core.channels import channel_identity
from ..core import reactions
from ..core.channels import channel_identity, split_channel_sender
from ..core.connection import Unsubscribe
from ..core.events import EventKind, MeshEvent
from ..core.models import (
@@ -49,6 +52,29 @@ _AWAITING_CAP = 16
#: Called with a message when its ack arrives after the send stopped its wait for the ack.
LateAck = Callable[[ChatMessage], None]
#: The highest try number that a direct message can carry. The protocol keeps it in two bits.
_LAST_ATTEMPT = 3
def _stamp() -> tuple[int, datetime]:
"""The timestamp for a message that MeshTerm is about to send, and the same instant as a date.
MeshTerm gives the timestamp to the device, and does not let the library read the clock
a second time. Thus the stored time of the message is exactly the time that the
recipients see. A reaction names its target by that time (refer to
:mod:`~meshterm.core.reactions`), so the two must agree to the second.
"""
seconds = int(time.time())
return seconds, datetime.fromtimestamp(seconds, tz=timezone.utc)
def _is_reaction(message: Message) -> bool:
"""Whether an inbound message is a reaction. A room post never is one (rooms have none)."""
if message.is_post:
return False
body = split_channel_sender(message.text)[1] if message.is_channel else message.text
return reactions.parse(body.strip(" \t"), channel=message.is_channel) is not None
def _fallback_channel_id(idx: int | None) -> str:
"""The identity of a channel that MeshTerm cannot read.
@@ -103,6 +129,9 @@ class ChatService:
# that wants to know when its ack arrives (refer to :meth:`_settle_ack`).
self._awaiting: OrderedDict[str, tuple[ChatMessage, LateAck | None]] = OrderedDict()
self._ack_unsubscribe: Unsubscribe | None = None
# The next try number of each direct message that was sent (by its row id). A
# retry with ^R sends the same message again: the same timestamp, the next number.
self._next_attempt: dict[int, int] = {}
@property
def active(self) -> bool:
@@ -408,6 +437,9 @@ class ChatService:
that comes from the slot.
* A muted channel (refer to :class:`~meshterm.core.mute_store.MuteStore`). This is
the one suppression that the user asks for, not one that the device implies.
* A reaction (refer to :mod:`~meshterm.core.reactions`). The transcript shows it on
the message that it reacts to, not as a new message. Thus a badge for it would
point at nothing new.
The method asks the device through the session cache. Thus this rule costs nothing
for each message, more than the channel read that the identity already made.
@@ -422,6 +454,8 @@ class ChatService:
Returns:
``True`` to increase the unread count of the conversation.
"""
if _is_reaction(message):
return False
if message.is_channel:
store = getattr(self._ctx, "mute_store", None)
if store is not None and store.is_muted(channel_id):
@@ -481,7 +515,9 @@ class ChatService:
return is_direct_messageable(contact.node_type)
return False
async def _deliver_direct(self, contact: Contact, text: str) -> tuple[bool, list[str]]:
async def _deliver_direct(
self, contact: Contact, text: str, *, timestamp: int, first_attempt: int = 0
) -> tuple[bool, list[str], int]:
"""Transmit a direct message, and try again softly until it is acknowledged.
One logical send makes one first transmission, plus up to
@@ -496,13 +532,22 @@ class ChatService:
stores the message as unacknowledged. Then Ctrl-R in the chat screen is the retry
that the user starts, in addition to these automatic retries.
Each try carries the same timestamp and the next try number, as the protocol
specifies. Thus the recipient sees one message with one timestamp, whichever try
arrives. A reaction to it names that timestamp. The try number makes each packet
different, so a repeater does not discard a retry as a copy of the first packet. The
number stops at 3 (the protocol has two bits for it).
Args:
contact: The recipient.
text: The message body.
timestamp: The timestamp of the message, the same for each try.
first_attempt: The try number of the first transmission.
Returns:
Whether an attempt was acknowledged in its wait, and the ack code of each
attempt. A late ack can still carry any of these codes (the device keeps them).
Whether an attempt was acknowledged in its wait, the ack code of each attempt (a
late ack can still carry any of these codes, because the device keeps them),
and the try number for the next retry.
Raises:
Exception: A hard send failure (the companion rejects the send) goes to the
@@ -512,12 +557,16 @@ class ChatService:
device = await self._ctx.device()
attempts = max(0, self._ctx.preferences.direct_message_soft_retries) + 1
codes: list[str] = []
number = first_attempt
for attempt in range(1, attempts + 1):
delivery: Delivery = await device.send_direct_message(contact, text)
delivery: Delivery = await device.send_direct_message(
contact, text, timestamp=timestamp, attempt=min(number, _LAST_ATTEMPT)
)
number += 1
if delivery.code:
codes.append(delivery.code)
if delivery.acked:
return True, codes
return True, codes, number
if attempt < attempts:
self._ctx.log.debug(
"chat: DM to %s unacked on try %d/%d; soft-retrying",
@@ -525,7 +574,7 @@ class ChatService:
attempt,
attempts,
)
return False, codes
return False, codes, number
async def send_direct(
self, contact: Contact, text: str, *, on_late_ack: LateAck | None = None
@@ -549,7 +598,8 @@ class ChatService:
The stored outbound :class:`ChatMessage`. Its ``acked`` shows whether a
delivery acknowledgement arrived in the wait, until now.
"""
acked, codes = await self._deliver_direct(contact, text)
seconds, sent_at = _stamp()
acked, codes, number = await self._deliver_direct(contact, text, timestamp=seconds)
chat = ChatMessage(
text=text,
outbound=True,
@@ -557,10 +607,11 @@ class ChatService:
peer=contact.key_prefix or contact.public_key[:12] or None,
peer_name=contact.name,
acked=acked,
created_at=utcnow(),
created_at=sent_at,
)
chat.row_id = self._ctx.repo.record_chat_message(chat, run_id=self._run_id)
if not acked:
self._next_attempt[chat.row_id] = number
self._await_ack(chat, codes, on_late_ack)
return chat
@@ -583,7 +634,13 @@ class ChatService:
Returns:
The same ``message``, with :attr:`~ChatMessage.acked` updated.
"""
acked, codes = await self._deliver_direct(contact, message.text)
seconds = int(message.created_at.timestamp())
first = self._next_attempt.get(message.row_id, 1) if message.row_id is not None else 1
acked, codes, number = await self._deliver_direct(
contact, message.text, timestamp=seconds, first_attempt=first
)
if message.row_id is not None:
self._next_attempt[message.row_id] = number
if message.acked:
return message # a late ack for an earlier attempt settled it during this send
message.acked = acked
@@ -756,7 +813,8 @@ class ChatService:
device = await self._ctx.device()
channel_id = await self.channel_id_for(index)
asked = scope if scope is not None else self.channel_scope(channel_id)
await device.send_channel_in_scope(index, text, asked)
seconds, sent_at = _stamp()
await device.send_channel_in_scope(index, text, asked, timestamp=seconds)
chat = ChatMessage(
text=text,
outbound=True,
@@ -764,7 +822,7 @@ class ChatService:
channel_id=channel_id,
channel_idx=index,
peer_name=label,
created_at=utcnow(),
created_at=sent_at,
scope=await self._sent_under(asked),
)
chat.row_id = self._ctx.repo.record_chat_message(chat, run_id=self._run_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.
+25 -3
View File
@@ -27,7 +27,7 @@ from rich.cells import cell_len
from rich.text import Text
from ..context import AppContext
from ..core import exitcodes
from ..core import exitcodes, reactions
from ..core.channels import (
CHANNEL_SLOT_PROBE_CAP,
DEFAULT_PUBLIC_SECRET,
@@ -60,7 +60,7 @@ from ..ui.fields import ChannelRef, NodeRef
from ..ui.menus import Lane, column_header, fit_cells, section_heading
from ..ui.pathline import SELF_GLYPH, PathHop
from ..ui.termfont import powerline_enabled
from ..ui.theme import name_style
from ..ui.theme import emoticon, name_style
from ..ui.tui import Choice, DeleteRequest, Separator
from ..ui.widgets import (
DESTINATION,
@@ -1047,7 +1047,7 @@ def _preview_text(
was cut before to approximately the visible lane has nothing more to show. The list
cuts it at the right edge, the same as any other row, and the scroll goes past that.
"""
body_raw = last.text.replace("\n", " ")
body_raw = _reaction_line(last) or last.text.replace("\n", " ")
text = Text()
if last.outbound:
_append_author(text, PathHop(SELF_GLYPH, you=True), Text("you: ", style="accent"))
@@ -1071,6 +1071,28 @@ def _preview_text(
return text
def _reaction_line(last: ChatMessage) -> str | None:
"""The preview body of a reaction, ``reacted 👍``, or ``None`` for a usual message.
A reaction is a message on the wire, and it can be the latest message of a
conversation. Its raw body (a mention, a newline, and a hash) means nothing to a person.
The name of the node keeps its place in front of the body, as for each message.
"""
if last.author is not None:
return None # a room post: rooms have no reactions
body = (
split_channel_sender(last.text)[1] if last.is_channel and not last.outbound else last.text
)
reaction = reactions.parse(body.strip(" \t"), channel=last.is_channel)
if reaction is None:
return None
line = f"reacted {emoticon(reaction.emoji)}"
if last.is_channel and not last.outbound:
name = split_channel_sender(last.text)[0]
return f"{name}: {line}" if name else line
return line
def _append_author(text: Text, hop: PathHop, byline: Text) -> None:
"""Append the author of a preview: an origin chip where chips are drawn, else ``byline``.
+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
+290 -22
View File
@@ -18,14 +18,16 @@ from __future__ import annotations
import asyncio
import re
import unicodedata
from collections.abc import Awaitable, Callable
from functools import lru_cache
from typing import TYPE_CHECKING, Any
from rich.cells import cell_len
from rich.console import Group, RenderableType
from rich.text import Text
from ..core import links
from ..core import links, reactions
from ..core.channels import MENTION, split_channel_sender
from ..core.connection import ContactNotOnDeviceError
from ..core.events import EventKind, MeshEvent
@@ -34,7 +36,7 @@ from ..core.regions import WILDCARD
from ..platforms import Platform, on_platform
from .pathline import PATH_INK
from .qr import QrScreen, qr_strip
from .theme import snr_style
from .theme import emoticon, name_style, snr_style
from .tui.prompt import (
CHANNEL_BYTE_LIMIT,
DM_BYTE_LIMIT,
@@ -74,12 +76,18 @@ _URL_GUTTER = 2
#: only on the PicoCalc). Bound for each platform, never asked at each frame.
_URL_CODES = False
#: The mark in front of the count of a reaction. Where the font has no emoji, an emoticon
#: such as ``+1`` or ``<3`` stands in for the emoji, and a bare count after it reads as part
#: of it (``+1 3``). There, the count is ``x3``. Bound for each platform.
_COUNT_MARK = ""
@on_platform
def _bind(platform: Platform) -> None:
"""Hang codes under URLs where the platform draws them (now and at each switch)."""
global _URL_CODES
"""Bind the choices of the transcript that depend on the platform (now and at each switch)."""
global _URL_CODES, _COUNT_MARK
_URL_CODES = platform.url_codes
_COUNT_MARK = "" if platform.emoji else "x"
#: The maximum width of a footer hint, in cells (refer to the UX standards).
@@ -113,9 +121,15 @@ class ChatScreen(Screen):
of the selected message. A path belongs to one message, so when no message is
selected, there is nothing to show. ^U shows the links of the selected message as QR
codes, on the share screen. ^L opens the map on the node of the author of the selected
message, when that node has a location. When no message is selected, Esc leaves the
message, when that node has a location. ^E reacts to the selected message with an emoji
(refer to :mod:`~meshterm.core.reactions`). When no message is selected, Esc leaves the
chat.
A reaction is a message on the wire, but the transcript does not show it as one. It shows
each reaction under the message that it reacts to: the emoji and a count, or the names of
the nodes that reacted while that message is selected. A room has no reactions, so a
room shows each message as it arrived.
A channel with a send scope shows its region in the title (``#ops · yul``). There, ^R
sends a message again *unscoped*: the selected message, or the newest when no message is
selected. This is the solution for a scoped message that no repeater in range relays.
@@ -148,6 +162,11 @@ class ChatScreen(Screen):
same size, so there the chip only shows one code alone. ``Map`` is lit while the
author of the selection has a location (refer to :meth:`_locate_target`).
``React`` (^E) is on the Shift half of F2, behind ``URL→QR``. A channel and a DM have
it, and a room does not. Thus it goes on the Shift bank, the bank for a verb that not
all three screens share. A room puts ``Log in`` there instead. ``React`` is lit while
the selection is a message that a reaction can name (refer to :meth:`_can_react`).
``Author`` (^O, the node page of the author of the selection) is the fourth shared
verb, and the plain bank has only three slots. Thus it is on the Shift half of F3,
behind ``Map``, the other verb about the author. It is lit while the author has a
@@ -193,7 +212,14 @@ class ChatScreen(Screen):
enabled=self._selected is not None and self._paths is not None,
opp_enabled=self._retry_target() is not None,
)
lane[1] = FPair("URL→QR", "url_code", enabled=bool(self._picked_urls()))
react = ("React", "react") if self._fold is not None else ("", "")
lane[1] = FPair(
"URL→QR",
"url_code",
*react,
enabled=bool(self._picked_urls()),
opp_enabled=self._can_react(),
)
if self._locator is not None or self._pages is not None:
where = ("Map", "locate") if self._locator is not None else ("", "")
who = ("Author", "author_page") if self._pages is not None else ("", "")
@@ -222,6 +248,8 @@ class ChatScreen(Screen):
resend_unscoped: Callable[[ChatMessage], Awaitable[ChatMessage | None]] | None = None,
locator: Locator | None = None,
pages: ContactPages | None = None,
self_name: str | None = None,
reaction_set: list[str] | None = None,
) -> None:
"""Build the chat screen.
@@ -259,6 +287,11 @@ class ChatScreen(Screen):
pages: The contacts, for ^O: the node page of the author of the selected
message (refer to :class:`~meshterm.ui.node_detail_screen.ContactPages`).
With ``None``, ^O does nothing.
self_name: The name of our node. A channel reaction names the author of its
target, and our node is the author of each message that we sent. With
``None``, our channel messages get no reactions.
reaction_set: The emoji that ^E offers, in order. ``None`` offers
:data:`~meshterm.core.reactions.DEFAULT_REACTIONS`.
"""
super().__init__()
self._scope = scope if conversation.is_channel else None
@@ -281,7 +314,15 @@ class ChatScreen(Screen):
if conversation.is_channel or contact is None
else (contact.public_key or contact.key_prefix or "")
)
self._messages = list(messages)
# The reactions of the conversation, folded onto their targets. A room has none.
self._fold = (
None
if conversation.is_room
else reactions.ReactionFold(channel=conversation.is_channel, self_name=self_name)
)
self._reaction_set = reaction_set or reactions.DEFAULT_REACTIONS.split()
self._peer_label = conversation.label
self._messages = [m for m in messages if not self._folds(m)]
self._send = send
self._resend = resend
self._resend_unscoped = resend_unscoped if conversation.is_channel else None
@@ -305,6 +346,7 @@ class ChatScreen(Screen):
self._codes_open = False # one share screen of links at a time
self._paste_open = False # one paste-confirm dialog at a time
self._resend_open = False # one resend-unscoped confirm at a time
self._react_open = False # one reaction dialog at a time
# The selection: the index of the highlighted message (or None when the compose
# line has the focus), and the body line where it rendered, so that the frame keeps
# it visible.
@@ -348,6 +390,7 @@ class ChatScreen(Screen):
code = "^U QR" if self._picked_urls() else ""
where = "^L map" if self._locate_target() is not None else ""
who = "^O author" if self._author_page() is not None else ""
react = "^E react" if self._can_react() else ""
if self._is_channel and self._retry_target() is not None:
# A selected scoped message of ours can be sent again unscoped, and the hint
# must name that key. The "(@mention)" of the reply and ^End are removed to
@@ -362,11 +405,12 @@ class ChatScreen(Screen):
code,
where,
who,
react,
"^R resend unscoped",
"↑↓ pick",
"Esc cancel",
],
spare=("↑↓ pick", "^P paths", "^O author"),
spare=("↑↓ pick", "^P paths", "^O author", "^E react"),
)
if self._replies:
# With all the keys of a selection, the long words do not fit in 72 cells.
@@ -378,6 +422,7 @@ class ChatScreen(Screen):
code,
where,
who,
react,
"↑↓ pick",
"^End/Esc cancel",
],
@@ -385,13 +430,15 @@ class ChatScreen(Screen):
)
if cell_len(hint) <= _HINT_CELLS:
return hint
# ^E goes after ↑↓ when even the short words do not fit. The keys of the
# selection that were here before ^E keep their places.
return _fit_hint(
["Enter reply", "^P paths", code, where, who, "↑↓ pick", "Esc cancel"],
spare=("↑↓ pick",),
["Enter reply", "^P paths", code, where, who, react, "↑↓ pick", "Esc cancel"],
spare=("↑↓ pick", "^E react"),
)
return _fit_hint(
["Enter paths", code, where, who, "↑↓ pick", "^End/Esc cancel"],
spare=("↑↓ pick",),
["Enter paths", code, where, who, react, "↑↓ pick", "^End/Esc cancel"],
spare=("↑↓ pick", "^E react"),
)
if self._retry_target() is not None:
if self._is_channel:
@@ -402,7 +449,14 @@ class ChatScreen(Screen):
# --- live updates --------------------------------------------------------
def append(self, message: ChatMessage) -> None:
"""Append an inbound message to the transcript and paint."""
"""Append an inbound message to the transcript and paint.
A reaction goes onto the message that it reacts to, and the transcript does not
move for it.
"""
if self._folds(message):
self._session.invalidate()
return
self._messages.append(message)
# Do not pull the viewport to the tail while the user selects a message to reply to.
if self._selected is None:
@@ -549,7 +603,8 @@ class ChatScreen(Screen):
tick),
* it is the selected message (the selection sets its style, not the message), or
* its identity or ``acked`` is different from the cache (it was appended, it was
replaced in place when a pending bubble resolved, or it was removed).
replaced in place when a pending bubble resolved, or it was removed), or
* its number of reactions is different from the cache (a reaction arrived).
Most importantly, this works for each message. It is not a cached prefix that
stops at the first such row. Thus a message stays cheap to draw again, however far
@@ -565,11 +620,12 @@ class ChatScreen(Screen):
cached_ids = cache["ids"]
cached_acked = cache["acked"]
cached_selected = cache["selected"]
cached_reacts = cache["reacts"]
cached_boundaries = cache["boundaries"]
cached_lines = cache["lines"]
cache_count = len(cached_ids)
else:
cached_ids = cached_acked = cached_selected = cached_boundaries = ()
cached_ids = cached_acked = cached_selected = cached_reacts = cached_boundaries = ()
cached_lines = []
cache_count = 0
@@ -583,6 +639,7 @@ class ChatScreen(Screen):
ids: list[int] = []
ackeds: list[bool | None] = []
selecteds: list[bool] = []
reacts: list[int] = []
boundaries: list[int] = []
prev_group: tuple[bool, str] | None = None
prev_day = None
@@ -598,6 +655,7 @@ class ChatScreen(Screen):
new_day = day != prev_day
selected = idx == self._selected
pending = message.acked is None and message.outbound and not message.is_channel
react_count = self._fold.count(message) if self._fold is not None else 0
reusable = (
not selected
and not pending
@@ -605,6 +663,7 @@ class ChatScreen(Screen):
and not cached_selected[idx]
and cached_ids[idx] == id(message)
and cached_acked[idx] == message.acked
and cached_reacts[idx] == react_count
)
if reusable:
start = cached_boundaries[idx - 1] if idx > 0 else 0
@@ -646,6 +705,7 @@ class ChatScreen(Screen):
ids.append(id(message))
ackeds.append(message.acked)
selecteds.append(selected)
reacts.append(react_count)
boundaries.append(len(lines))
self._render_cache = {
@@ -653,6 +713,7 @@ class ChatScreen(Screen):
"ids": ids,
"acked": ackeds,
"selected": selecteds,
"reacts": reacts,
"lines": lines,
"boundaries": boundaries,
}
@@ -741,8 +802,51 @@ class ChatScreen(Screen):
urls = links.urls(body) if _URL_CODES else []
if urls:
lines += render_lines(qr_strip(urls, width, indent=prefix.cell_len), width)
badges = self._fold.badges(message) if self._fold is not None else []
if badges:
row, runs = self._badge_text(badges, selected=selected)
lines += render_hanging(
Text(" " * prefix.cell_len), row, width, indent=prefix.cell_len, whole=runs
)
return lines
def _badge_text(
self, badges: list[reactions.Badge], *, selected: bool
) -> tuple[Text, list[tuple[int, int]]]:
"""The reactions on one message, as one line under its body, and the runs to keep whole.
Each emoji has its count after it. The count is ``accent`` when our node is one of
the nodes that reacted, else ``muted``. On the selected message, the names of the
nodes take the place of the count, each in the hue of its node, and our node is the
white ``you``. Where the font has no emoji, an emoticon stands in for each default
reaction (refer to :func:`~meshterm.ui.theme.emoticon`).
"""
text = Text()
runs: list[tuple[int, int]] = []
for index, badge in enumerate(badges):
if index:
text.append(" ")
start = len(text)
text.append(emoticon(badge.emoji))
text.append(" ")
if not selected:
count = f"{_COUNT_MARK}{len(badge.reactors)}"
text.append(count, style="accent" if badge.ours else "muted")
runs.append((start, len(text)))
continue
for number, reactor in enumerate(badge.reactors):
if number:
text.append(", ", style="muted")
if reactor.ours:
text.append("you", style="you")
else:
name = reactor.name or self._peer_label
key = self._peer_key if reactor.name is None else self._sender_key(name)
text.append(name, style=name_style(name, key or None))
if not number:
runs.append((start, len(text))) # the emoji never ends a line alone
return text, runs
def _body_text(self, body: str, message: ChatMessage, *, selected: bool) -> Text:
"""Build the styled body of a message: mentions in colour, then the glyphs at the end."""
text = self._render_mentions(body, selected=selected)
@@ -872,10 +976,10 @@ class ChatScreen(Screen):
Ctrl+PgUp/PgDn (day dividers), and the viewport moves with it. Enter on a selected
message starts a reply with an ``@mention`` in a channel, and opens the delivery
paths in a direct chat. In the two types of chat, ^P opens the paths of the
selected message, ^U the QR codes of its links, and ^L the map on its author. The
three do nothing when no message is selected. ^End (or a move past the newest
message) goes back to the compose line. The first Esc clears the selection, and the
second Esc leaves the screen.
selected message, ^U the QR codes of its links, ^L the map on its author, and ^E the
reaction dialog. The four do nothing when no message is selected. ^End (or a move
past the newest message) goes back to the compose line. The first Esc clears the
selection, and the second Esc leaves the screen.
"""
if action == "enter":
if self._selected is not None:
@@ -895,6 +999,8 @@ class ChatScreen(Screen):
self._open_map()
elif action == "author_page":
self._open_author_page()
elif action == "react":
self._begin_react()
elif action == "retry":
if self._is_channel:
self._begin_resend_unscoped()
@@ -1224,7 +1330,7 @@ class ChatScreen(Screen):
try:
message = await self._send(text)
if message is not None:
self._messages.append(message)
self._admit(message)
self._status = ""
except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open
self._status = f"send failed: {exc}"
@@ -1382,7 +1488,7 @@ class ChatScreen(Screen):
try:
message = await self._resend_unscoped(target)
if message is not None:
self._messages.append(message)
self._admit(message)
self._status = ""
except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open
self._status = f"resend failed: {exc}"
@@ -1409,12 +1515,126 @@ class ChatScreen(Screen):
self._session.invalidate()
def _swap(self, old: ChatMessage, new: ChatMessage) -> None:
"""Replace an optimistic bubble with its stored message, in place."""
"""Replace an optimistic bubble with its stored message, in place.
The stored message is the one that a reaction can name, because it has the
timestamp that went out. Thus the fold takes it here, and not the bubble.
"""
if self._folds(new):
self._discard(old)
return
try:
self._messages[self._messages.index(old)] = new
except ValueError: # pragma: no cover - the bubble is always still there
self._messages.append(new)
def _folds(self, message: ChatMessage) -> bool:
"""Give ``message`` to the fold. ``True`` for a reaction, which the transcript omits."""
return self._fold is not None and self._fold.take(message)
def _admit(self, message: ChatMessage) -> None:
"""Append a message that we sent, unless it is a reaction (the fold shows that one)."""
if not self._folds(message):
self._messages.append(message)
# --- reactions -------------------------------------------------------------
def _react_target(self) -> tuple[ChatMessage, str | None, str, int] | None:
"""The selected message, and the author, body, and timestamp that a reaction names it by.
``None`` when nothing is selected, or when a reaction cannot name the selection: a
room, a message that is still in flight, or a channel message with no author (a
line with no ``Name: `` prefix, or one of ours when our name is not known).
"""
if self._fold is None or self._selected is None or not self._messages:
return None
message = self._messages[min(self._selected, len(self._messages) - 1)]
named = self._fold.target_of(message)
if named is None:
return None
author, body, timestamp = named
if self._is_channel and not author:
return None
return message, author, body, timestamp
def _can_react(self) -> bool:
"""Whether ^E can start a reaction: a selection that a reaction can name, and no send."""
return not self._sending and self._composing and self._react_target() is not None
def _begin_react(self) -> None:
"""Ask which emoji, then send the reaction to the selected message (^E).
The dialog is a question, so it floats as a box over the chat. The emoji of the
``reaction_emoji`` preference come first, then each other emoji (refer to
:func:`emoji_catalog`). Each row has the name of its emoji, so a typed word such as
``fire`` or ``heart`` filters the list to the emoji that it names. A row that our
node already sent to this message says so. A second copy of it would only count one
time, so the selection of that row sends nothing. The reaction goes out as a usual
message in the conversation. When it is stored, the fold puts it under its target
(refer to :meth:`_admit`).
"""
target = self._react_target()
if target is None or self._sending or self._react_open or self._fold is None:
return
message, author, body, timestamp = target
fold = self._fold
self._react_open = True
sender, _ = self._sender_and_body(message)
who = "your message" if message.outbound else sender
async def run() -> None:
from .menus import menu_rows, section_heading
def row(emoji: str) -> tuple[str, str, str]:
name = reactions.emoji_name(emoji)
sent = fold.reacted(message, emoji)
return emoticon(emoji), f"{name} · sent" if sent else name, emoji
mine = list(self._reaction_set)
others = [emoji for emoji in emoji_catalog() if emoji not in mine]
rows = menu_rows([row(emoji) for emoji in mine + others])
items = [
section_heading("Your reactions"),
*rows[: len(mine)],
section_heading("All emoji"),
*rows[len(mine) :],
]
try:
emoji = await self._session.select(
f"React — {who}", items, filterable=True, floating=True
)
finally:
self._react_open = False
if not emoji or self._sending:
self._session.invalidate()
return
if fold.reacted(message, emoji):
self._status = "you already reacted with that"
self._session.invalidate()
return
if self._is_channel:
text = reactions.channel_text(emoji, author or "", body, timestamp)
else:
text = reactions.direct_text(emoji, body, timestamp)
self._sending = True
self._status = "reacting…"
self._session.invalidate()
try:
sent_message = await self._send(text)
if sent_message is not None:
self._admit(sent_message)
no_ack = not self._is_channel and sent_message.acked is False
self._status = "reaction sent · no ack" if no_ack else ""
else:
self._status = ""
except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open
self._status = f"reaction failed: {exc}"
finally:
self._sending = False
self._session.invalidate()
self._spawn(run())
def _discard(self, message: ChatMessage) -> None:
"""Remove an optimistic bubble that never became a real message."""
try:
@@ -1423,6 +1643,47 @@ class ChatScreen(Screen):
pass
#: The blocks of Unicode where an emoji with the emoji presentation can be. A wide symbol
#: between them is a CJK symbol or radical, not an emoji.
_EMOJI_BLOCKS = ((0x2000, 0x2E7F), (0x1F000, 0x1FAFF))
#: The I Ching symbols, which Unicode 16 draws wide. They are symbols, not emoji: the
#: firmware and the other apps never offer them.
_NOT_EMOJI = frozenset(range(0x268A, 0x2690)) | frozenset(range(0x2630, 0x2638))
@lru_cache(maxsize=1)
def emoji_catalog() -> tuple[str, ...]:
"""Each single emoji that the reaction dialog offers, in the order of Unicode.
The standard library has no list of emoji. Thus the catalog takes two sets. The first is
each symbol in the emoji blocks that Unicode draws wide by default (``🔥``, ``⌚``). The
second is the emoji that are text by default (``❤``), which the table of Rich lists.
Each of these gets the variation selector that asks for the emoji form (``❤️``), as the
apps send it. A flag, a skin tone, and a joined sequence are not in the catalog, but a
reaction with one still shows when it arrives. The names come from the Unicode database
of the Python that runs MeshTerm, so a newer Python offers newer emoji.
"""
from rich._unicode_data import load
found: dict[int, str] = {}
for first, last in _EMOJI_BLOCKS:
for cp in range(first, last + 1):
char = chr(cp)
if cp in _NOT_EMOJI:
continue
if unicodedata.category(char) == "So" and unicodedata.east_asian_width(char) == "W":
found[cp] = char
for char in load("auto").narrow_to_wide:
if not char.isascii() and len(char) == 1:
found.setdefault(ord(char), char + "\ufe0f")
return tuple(
emoji
for _, emoji in sorted(found.items())
if not reactions._is_regional(emoji[0]) and reactions.is_reaction_emoji(emoji)
)
async def open_chat(ctx: AppContext, conversation: Conversation) -> int:
"""Open the live chat screen for ``conversation`` and run it until the user closes it.
@@ -1504,6 +1765,11 @@ async def open_chat(ctx: AppContext, conversation: Conversation) -> int:
from .node_detail_screen import ContactPages
paths = await _make_paths_presenter(ctx, conversation, device)
self_name: str | None = None
try:
self_name = str((await ctx.devstate.self_info()).get("name") or "") or None
except Exception: # noqa: BLE001 - with no name, our channel messages get no reactions
self_name = None
screen = ChatScreen(
conversation,
history,
@@ -1517,6 +1783,8 @@ async def open_chat(ctx: AppContext, conversation: Conversation) -> int:
resend_unscoped=resend_unscoped,
locator=await Locator.gather(ctx),
pages=await ContactPages.gather(ctx),
self_name=self_name,
reaction_set=reactions.reaction_set(ctx.preferences.reaction_emoji),
)
ctx.chat.set_active(conversation.key)
+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)
+33 -1
View File
@@ -825,6 +825,36 @@ def _glyph_identity(icon: str) -> str:
return icon
#: The emoticons that stand in for the default reactions where the font has no emoji (the
#: PicoCalc and the Cardputer). Each default reaction is two cells wide, and so is each
#: emoticon. Thus a badge keeps its width. A reaction that is not here goes through the fold
#: at the render boundary, as each other emoji does.
# fmt: off
_EMOTICONS: dict[str, str] = {
"👍": "+1", "👎": "-1", "❤️": "<3", "❤": "<3", "😂": ":D", "😮": ":O", "😢": ":(",
}
# fmt: on
def emoticon(emoji: str) -> str:
"""The rendering of a reaction on the platform: the emoji itself, or its emoticon.
Args:
emoji: The emoji of a reaction (one glyph, refer to
:func:`~meshterm.core.reactions.is_reaction_emoji`).
Returns:
The emoji on a platform that draws emoji. Else its emoticon from
:data:`_EMOTICONS`, or the emoji unchanged when it has none.
"""
return _emoticon_impl(emoji)
def _emoticon_compact(emoji: str) -> str:
"""A platform with no emoji: the emoticon of a reaction, when it has one."""
return _EMOTICONS.get(emoji, emoji)
def _glyph_compact(icon: str) -> str:
"""The PicoCalc: icons become their single-cell console-font glyph."""
return _GLYPH_MAP.get(icon, icon)
@@ -1099,6 +1129,7 @@ def snr_style(snr: float | None) -> str:
_ACTIVE_THEME: Theme = MESH_THEME
_node_impl: Callable[[str], str] = _node_style_spectrum
_glyph_impl: Callable[[str], str] = _glyph_identity
_emoticon_impl: Callable[[str], str] = _glyph_identity
_fold_impl: Callable[[str], str] = _no_fold
#: The map from a marker colour to RGB. :func:`mark_rgb` fills it, and a platform switch
@@ -1109,11 +1140,12 @@ _STYLE_RGB: dict[str, tuple[int, int, int]] = {}
@on_platform
def _bind(platform: Platform) -> None:
"""Bind the choices of the theme that depend on the platform (runs now and at each switch)."""
global _ACTIVE_THEME, _node_impl, _glyph_impl, _fold_impl, _FOLD_TABLE
global _ACTIVE_THEME, _node_impl, _glyph_impl, _emoticon_impl, _fold_impl, _FOLD_TABLE
global _FOLD_FONT, _FOLD_ALLOWED, _FOLD_QUANTIZE
_ACTIVE_THEME = MESH_THEME if platform.truecolor else MESH_THEME_16
_node_impl = _node_style_spectrum if platform.truecolor else _node_style_quantized
_glyph_impl = _glyph_identity if platform.emoji else _glyph_compact
_emoticon_impl = _glyph_identity if platform.emoji else _emoticon_compact
font = FONTS.get(platform.font)
if font is not None:
_FOLD_FONT = font
+3
View File
@@ -102,6 +102,9 @@ from .spinner import spinner_interval
#: ``IXON``/``IXOFF``, so ^Q is not XON here.
_CTRL_LETTER_CHORDS: dict[str, str] = {
"c": "quit",
# ^E reacts to the selected message with an emoji. The raw mode of prompt_toolkit has no
# line editing, so ^E is not "end of line" here (the End key moves the caret).
"e": "react",
# ^G logs in to the room that a room screen shows. In the input, ^G is not a special key:
# it is the bell only in output, and no terminal or multiplexer takes it.
"g": "login",
+41 -3
View File
@@ -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)
+7 -2
View File
@@ -1673,14 +1673,19 @@ class _ScriptedDevice:
def __init__(self, acks: list) -> None:
self._acks = list(acks)
self.sent: list[str] = []
# The timestamp and the try number of each send, in order.
self.stamps: list[tuple[int | None, int]] = []
async def connect(self) -> None: # for _StubContext.device()
return None
async def send_direct_message(self, contact: Contact, text: str):
async def send_direct_message(
self, contact: Contact, text: str, *, timestamp: int | None = None, attempt: int = 0
):
from meshterm.core.models import Delivery
self.sent.append(text)
self.stamps.append((timestamp, attempt))
ack = self._acks.pop(0) if self._acks else None
return Delivery(f"{len(self.sent):08x}", ack)
@@ -2710,7 +2715,7 @@ class _AckingMeshCore:
if sub.event_type is EventType.ACK:
sub.callback(event)
async def send_msg(self, pub, text): # noqa: ANN001, ANN201
async def send_msg(self, pub, text, timestamp=None, attempt=0): # noqa: ANN001, ANN201
from types import SimpleNamespace
if self._during:
+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(
+9 -4
View File
@@ -244,7 +244,8 @@ def test_the_app_draws_no_text_default_emoji_its_vocabulary_does_not_name() -> N
screen, and does not name in the list, fails here. Without the test, the mark would get
a blank cell next to it with no warning. The test skips prose, because MeshTerm reads
docstrings and bare strings and does not draw them. It also skips the module that defines
the list.
the list. It also skips the vocabulary of reactions, which is content that other apps
send.
"""
from rich._unicode_data import load
@@ -254,13 +255,17 @@ def test_the_app_draws_no_text_default_emoji_its_vocabulary_does_not_name() -> N
if not char.isascii() and char not in cells._SINGLE_CELLS
}
# Inputs to the substitution table of the PicoCalc, which maps them to glyphs that its
# font has. A terminal that shows emoji never draws them, so they are not marks.
fold_inputs = {"↪", "✔", "✖"}
# font has. A terminal that shows emoji never draws them, so they are not marks. The
# heart is the key of an emoticon (theme._EMOTICONS), which only a handheld uses.
fold_inputs = {"↪", "✔", "✖", "❤"}
package = pathlib.Path(ew.__file__).parents[2]
drawn: set[str] = set()
for path in package.rglob("*.py"):
if path.name == "emoji_width.py":
# The reactions module holds the emoji that other apps send (the defaults of
# MeshCore One, the table of meshcore-open). These are content, not marks of the
# app, and each one that is text-default carries its variation selector.
if path.name in ("emoji_width.py", "reactions.py"):
continue
tree = ast.parse(path.read_text(encoding="utf-8"))
prose = {
+40
View File
@@ -735,6 +735,45 @@ def _chat_urls(cols: int, rows: int) -> Screen:
return ChatScreen(conv, messages, send=None, names={}, session=_GallerySession(cols, rows))
def _chat_reactions(cols: int, rows: int) -> Screen:
"""A selected channel message with reactions: the names of the nodes take the counts' place.
Five nodes react with three emoji, one of them ours. The selection draws the widest line
that reactions make, and it wraps at the indent of the body. On the PicoCalc, emoticons
stand in for the emoji that have one.
"""
from datetime import datetime, timezone
from meshterm.core import reactions
stamp = 1_760_000_000
at = datetime.fromtimestamp(stamp, tz=timezone.utc)
def react(name: str, emoji: str) -> ChatMessage:
body = reactions.channel_text(emoji, "Alice", "bridge on 3rd is open again", stamp)
return ChatMessage(text=f"{name}: {body}", is_channel=True, created_at=at)
conv = Conversation(label="Lakeside emergency", is_channel=True, channel_idx=2)
messages = [
ChatMessage(text="Alice: bridge on 3rd is open again", is_channel=True, created_at=at),
react("Hilltop-Repeater", "👍"),
react("Bob", "👍"),
react("Carol-Mobile", "❤️"),
react("Dave", "🔥"),
ChatMessage(
text=reactions.channel_text("👍", "Alice", "bridge on 3rd is open again", stamp),
outbound=True,
is_channel=True,
created_at=at,
),
]
screen = ChatScreen(
conv, messages, send=None, names={}, session=_GallerySession(cols, rows), self_name="Me"
)
screen.handle("up")
return screen
def _chat_long_url(cols: int, rows: int) -> Screen:
"""A picked message with a link that is wider than the body lane.
@@ -1738,6 +1777,7 @@ _ENTRIES: list[_Entry] = [
_Entry("chat", _chat),
_Entry("chat_channel_scoped", _chat_channel_scoped),
_Entry("chat_urls", _chat_urls),
_Entry("chat_reactions", _chat_reactions),
_Entry("chat_long_url", _chat_long_url),
_Entry("chat_picker", _chat_picker),
_Entry("room", _room),
+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
+436
View File
@@ -0,0 +1,436 @@
# SPDX-License-Identifier: Apache-2.0
"""Emoji reactions: the wire format of MeshCore One, the fold, the send path, and the chat.
The parse cases come from the tests of MeshCore One (``ReactionParserTests.swift`` and
``MeshCoreOpenReactionParserTests.swift``). Thus MeshTerm reads the same bodies as the app
whose format it shares. The hash test calculates the expected value a second way, with the
Base32 of the standard library, so that the test does not only repeat the code.
"""
from __future__ import annotations
import asyncio
import base64
import hashlib
import logging
from datetime import datetime, timezone
from pathlib import Path
import pytest
from meshterm.core import reactions
from meshterm.core.config import Settings
from meshterm.core.models import ChatMessage, Contact, Conversation, Delivery, Message
from meshterm.core.preferences import Preferences
from meshterm.core.reactions import MC1, MCO, MCO_V1, ReactionFold, parse
from meshterm.persistence.repository import Repository
from meshterm.services.chat_service import ChatService
from meshterm.ui.chat import ChatScreen
from tests.conftest import plain
_T = 1_704_067_200 # 2024-01-01 00:00:00 UTC
def _at(seconds: int) -> datetime:
return datetime.fromtimestamp(seconds, tz=timezone.utc)
# -- the hash -----------------------------------------------------------------------------
def test_the_hash_is_crockford_base32_of_the_first_five_bytes_of_sha256() -> None:
"""SHA-256 of the UTF-8 body and the little-endian timestamp, five bytes, Crockford Base32."""
digest = hashlib.sha256(b"Hello" + _T.to_bytes(4, "little")).digest()[:5]
rfc = base64.b32encode(digest).decode() # five bytes make exactly eight characters
expected = rfc.translate(
str.maketrans("ABCDEFGHIJKLMNOPQRSTUVWXYZ234567", "0123456789abcdefghjkmnpqrstvwxyz")
)
assert reactions.message_hash("Hello", _T) == expected
assert reactions.message_hash("Hello", _T + 1) != expected
assert reactions.message_hash("World", _T) != expected
def test_the_texts_that_meshterm_sends_name_their_target() -> None:
"""A channel reaction names the author first. A direct one has the emoji and the hash."""
digest = reactions.message_hash("hi all", _T)
assert reactions.channel_text("👍", "Alice", "hi all", _T) == f"@[Alice]👍\n{digest}"
assert reactions.direct_text("👍", "hi all", _T) == f"👍\n{digest}"
assert parse(reactions.channel_text("👍", "Alice", "hi all", _T), channel=True) == (
reactions.Reaction("👍", MC1, digest, author="Alice")
)
# -- the parse ----------------------------------------------------------------------------
@pytest.mark.parametrize(
("body", "emoji", "author", "digest"),
[
("@[AlphaNode]👍\n7f3a9c12", "👍", "AlphaNode", "7f3a9c12"),
("❤️@[BetaNode]\ne4d8b1a0", "❤️", "BetaNode", "e4d8b1a0"), # before 1.4.1
("@[Node]👍\nOOOOOOOO", "👍", "Node", "00000000"),
("@[Node]👍\niiiiiiii", "👍", "Node", "11111111"),
("@[Node]👍\nLLLLLLLL", "👍", "Node", "11111111"),
("@[Node:Alpha]👍\na1b2c3d4", "👍", "Node:Alpha", "a1b2c3d4"),
("@[Alice]👍\nreceived", "👍", "Alice", "rece1ved"),
("@[A]👍🏽\na1b2c3d4", "👍🏽", "A", "a1b2c3d4"),
("@[A]👨‍👩‍👧\na1b2c3d4", "👨‍👩‍👧", "A", "a1b2c3d4"),
("@[A]🇨🇦\na1b2c3d4", "🇨🇦", "A", "a1b2c3d4"),
],
)
def test_a_channel_reaction_parses(body: str, emoji: str, author: str, digest: str) -> None:
"""The two orders, the Crockford substitutions, a colon in a name, and each glyph shape."""
assert parse(body, channel=True) == reactions.Reaction(emoji, MC1, digest, author=author)
@pytest.mark.parametrize(
"body",
[
"Just a normal message",
"@[Node]👍",
"👍 [Node]\na1b2c3d4",
"👍@Node\na1b2c3d4",
"@[Node]👍\nabc",
"@[Node]👍\nuuuuuuuu",
"@[]👍\na1b2c3d4",
"@[Node]\na1b2c3d4",
"@[Alice]🎉 on my way\npassword",
"🎉 on my way@[Alice]\npassword",
"A@[Node]\na1b2c3d4",
],
)
def test_a_usual_channel_message_is_not_a_reaction(body: str) -> None:
"""A body that only starts with an emoji, or is short of a part, stays a message."""
assert parse(body, channel=True) is None
def test_a_direct_reaction_has_no_author() -> None:
"""A direct reaction is the emoji and the hash. The channel shape is not one."""
assert parse("👍\nb45pc4ek", channel=False) == reactions.Reaction("👍", MC1, "b45pc4ek")
assert parse("@[Node]👍\na1b2c3d4", channel=False) is None
assert parse("🎉 on my way\npassword", channel=False) is None
assert parse("1\na1b2c3d4", channel=False) is None
def test_the_reactions_of_meshcore_open_parse() -> None:
"""The v3 form indexes a table of 184 emoji. The v1 form carries the time and two hashes."""
assert len(reactions.MCO_EMOJI) == 184
assert parse("r:a1b2:00", channel=True) == reactions.Reaction("👍", MCO, "a1b2")
assert parse("r:0000:45", channel=False).emoji == "😶" # the last smiley
assert parse("r:0000:66", channel=False).emoji == "💪" # the last gesture
assert parse("r:0000:67", channel=False).emoji == "❤️" # the first heart
assert parse("r:0000:b7", channel=False).emoji == "🚀" # the last object
for bad in ("r:a1b2:b8", "r:A1B2:00", "r:a1b2:0A", "r:a1b:00", "x:a1b2:00", ""):
assert parse(bad, channel=True) is None
old = parse("r:1704067200000_12345_67890:👍", channel=True)
assert old == reactions.Reaction(
"👍", MCO_V1, "1704067200000_12345_67890", timestamp=_T, author_hash=12345, body_hash=67890
)
def test_the_dart_hash_is_thirty_bits_and_never_zero() -> None:
"""The ``String.hashCode`` of Dart: 1 for no input, and always under 2**30."""
assert reactions.dart_hash([]) == 1
value = reactions.dart_hash([ord(c) for c in "test string"])
assert 0 < value < 1 << 30
assert reactions.dart_hash([ord(c) for c in "hello"]) != reactions.dart_hash(
[ord(c) for c in "world"]
)
def test_the_reaction_set_keeps_only_single_emoji() -> None:
"""A word that is not one emoji is ignored. With nothing left, the default set is used."""
assert reactions.reaction_set("🔥 ok 👍 🔥 👍👍") == ["🔥", "👍"]
assert reactions.reaction_set("") == reactions.DEFAULT_REACTIONS.split()
assert reactions.reaction_set("abc") == reactions.DEFAULT_REACTIONS.split()
# -- the fold -----------------------------------------------------------------------------
def _channel(text: str, seconds: int = _T, *, outbound: bool = False) -> ChatMessage:
return ChatMessage(text=text, outbound=outbound, is_channel=True, created_at=_at(seconds))
def test_a_channel_reaction_folds_onto_its_target_once_for_each_node() -> None:
"""The target stays. The reaction leaves the transcript. A copy from the same node is one."""
fold = ReactionFold(channel=True, self_name="Me")
target = _channel("Alice: hi all")
assert fold.take(target) is False
react = reactions.channel_text("👍", "Alice", "hi all", _T)
assert fold.take(_channel(f"Bob: {react}", _T + 5)) is True
assert fold.take(_channel(f"Bob: {react}", _T + 6)) is True # a relayed copy
assert fold.take(_channel(f"Carol: {react}", _T + 7)) is True
assert fold.take(_channel(reactions.channel_text("❤️", "Alice", "hi all", _T), outbound=True))
badges = fold.badges(target)
assert [(b.emoji, [r.name for r in b.reactors]) for b in badges] == [
("👍", ["Bob", "Carol"]),
("❤️", [None]),
]
assert badges[1].ours and fold.reacted(target, "❤️") and not fold.reacted(target, "👍")
assert fold.count(target) == 3
def test_a_reaction_to_our_channel_message_names_our_node() -> None:
"""Our channel messages have no prefix in the history. Our name is their author."""
fold = ReactionFold(channel=True, self_name="Me")
ours = _channel("lunch?", outbound=True)
fold.take(ours)
fold.take(_channel("Bob: " + reactions.channel_text("😂", "Me", "lunch?", _T)))
assert [b.emoji for b in fold.badges(ours)] == ["😂"]
assert fold.target_of(ours) == ("Me", "lunch?", _T)
def test_a_reaction_that_arrives_first_waits_for_its_target() -> None:
"""On a mesh, a reaction can arrive before its target. It attaches when the target comes."""
fold = ReactionFold(channel=True)
assert fold.take(_channel("Bob: " + reactions.channel_text("👍", "Alice", "late", _T)))
target = _channel("Alice: late")
assert fold.take(target) is False
assert [b.emoji for b in fold.badges(target)] == ["👍"]
def test_the_author_tells_two_identical_messages_apart() -> None:
"""Two nodes that send the same body in the same second get their own reactions."""
fold = ReactionFold(channel=True)
alice, carol = _channel("Alice: ok"), _channel("Carol: ok")
fold.take(alice)
fold.take(carol)
fold.take(_channel("Bob: " + reactions.channel_text("👍", "Carol", "ok", _T)))
assert fold.badges(alice) == [] and len(fold.badges(carol)) == 1
def test_a_direct_reaction_folds_by_hash_alone() -> None:
"""In a direct exchange, the peer is the only other node. Its reactor has no name."""
fold = ReactionFold(channel=False)
ours = ChatMessage(text="see you", outbound=True, created_at=_at(_T))
fold.take(ours)
assert fold.take(ChatMessage(text=reactions.direct_text("👍", "see you", _T), peer="d4e5"))
(badge,) = fold.badges(ours)
assert badge.emoji == "👍" and badge.reactors == [reactions.Reactor(None)]
def test_a_reaction_of_meshcore_open_finds_its_target_near_in_time() -> None:
"""The short hash of v3 is compared with the messages within the window of the reaction."""
fold = ReactionFold(channel=True)
target = _channel("Alice: hello there")
fold.take(target)
digest = reactions.mco_hash(_T, "Alice", "hello there")
assert fold.take(_channel(f"Bob: r:{digest}:05", _T + 30))
assert [b.emoji for b in fold.badges(target)] == ["🔥"]
far = _channel("Carol: hello there", _T + 10_000)
fold.take(far)
late = reactions.mco_hash(_T + 10_000, "Carol", "hello there")
assert fold.take(_channel(f"Bob: r:{late}:05", _T)) # out of the window: no target
assert fold.badges(far) == []
# -- the send path ------------------------------------------------------------------------
class _Recorder:
"""A device that records the timestamp and the try number of each send."""
def __init__(self, acks: list[bool]) -> None:
self._acks = list(acks)
self.direct: list[tuple[str, int | None, int]] = []
self.channel: list[tuple[str, int | None]] = []
async def send_direct_message(self, contact, text, *, timestamp=None, attempt=0): # noqa: ANN001, ANN201
self.direct.append((text, timestamp, attempt))
acked = self._acks.pop(0) if self._acks else False
code = f"{len(self.direct):08x}"
return Delivery(code, object() if acked else None)
async def send_channel_in_scope(self, index, text, scope, *, timestamp=None): # noqa: ANN001, ANN201
self.channel.append((text, timestamp))
class _Ctx:
def __init__(self, device: _Recorder, repo: Repository) -> None:
self._device = device
self.repo = repo
self.log = logging.getLogger("test.reactions")
self.settings = Settings()
self.preferences = Preferences()
async def device(self) -> _Recorder:
return self._device
@pytest.fixture()
def repo(tmp_path: Path) -> Repository:
"""A Repository on a scratch database. It closes when the test ends."""
r = Repository(tmp_path / "reactions.db")
yield r
r.close()
_ALICE = Contact(name="Alice", public_key="d4e5" + "0" * 60, key_prefix="d4e5f6a7")
async def test_a_sent_message_stores_the_timestamp_that_went_out(repo: Repository) -> None:
"""The stored time is the wire time to the second. A reaction to it then names it right."""
device = _Recorder([True])
chat = ChatService(_Ctx(device, repo))
sent = await chat.send_channel(0, "hello")
assert device.channel[0][1] == int(sent.created_at.timestamp())
assert sent.created_at.microsecond == 0
dm = await chat.send_direct(_ALICE, "hey")
assert device.direct[0][1] == int(dm.created_at.timestamp())
stored = repo.recent_chat_messages(is_channel=False, peer=_ALICE.key_prefix)
assert stored[-1].created_at == dm.created_at
async def test_retries_keep_the_timestamp_and_count_the_try(repo: Repository) -> None:
"""Soft retries and ^R send one message: the same timestamp, the next try number, up to 3."""
device = _Recorder([False, False, False, False, False])
ctx = _Ctx(device, repo)
ctx.preferences.direct_message_soft_retries = 1
chat = ChatService(ctx)
sent = await chat.send_direct(_ALICE, "hey")
await chat.resend_direct(_ALICE, sent)
await chat.resend_direct(_ALICE, sent)
stamps = {timestamp for _, timestamp, _ in device.direct}
assert stamps == {int(sent.created_at.timestamp())}
assert [attempt for *_, attempt in device.direct] == [0, 1, 2, 3, 3, 3]
async def test_a_reaction_does_not_raise_the_unread_badge(repo: Repository) -> None:
"""A reaction shows on its target, so a badge for it would point at nothing new."""
chat = ChatService(_Ctx(_Recorder([]), repo))
react = Message(text=reactions.direct_text("👍", "hey", _T), sender="d4e5f6a7")
assert await chat._notifies(react, None) is False
channel = Message(
text="Bob: " + reactions.channel_text("👍", "Alice", "hi", _T), channel=0, is_channel=True
)
assert await chat._notifies(channel, "abc") is False
# -- the chat screen ----------------------------------------------------------------------
class _Session:
"""A session stand-in that answers the reaction dialog with ``answer``."""
def __init__(self, answer: str | None) -> None:
self.answer = answer
self.asked: list[str] = []
self.items: list = []
self.filterable: bool | None = None
def invalidate(self) -> None:
pass
async def select(self, title, items, *, filterable=True, **_kw): # noqa: ANN001, ANN003, ANN201
self.asked.append(title)
self.items, self.filterable = items, filterable
return self.answer
def _screen(messages: list[ChatMessage], session: _Session, sent: list[str]) -> ChatScreen:
async def send(text: str) -> ChatMessage:
sent.append(text)
return ChatMessage(text=text, outbound=True, is_channel=True, created_at=_at(_T + 60))
return ChatScreen(
Conversation(label="#public", is_channel=True, channel_idx=0),
messages,
send=send,
names={},
session=session,
self_name="Me",
)
def test_the_transcript_shows_reactions_under_their_target() -> None:
"""The reaction leaves the transcript. Its emoji and count show under the target."""
react = "Bob: " + reactions.channel_text("👍", "Alice", "hi all", _T)
screen = _screen([_channel("Alice: hi all"), _channel(react, _T + 5)], _Session(None), [])
assert len(screen._messages) == 1
body = plain(screen.render_body(60))
assert "👍 1" in body
assert reactions.message_hash("hi all", _T) not in body and "@[Alice]" not in body
screen.handle("up") # the selection names the nodes instead of the count
assert "👍 Bob" in plain(screen.render_body(60))
async def test_ctrl_e_sends_the_reaction_and_folds_our_copy() -> None:
"""^E asks which emoji, sends the text of MeshCore One, and shows our reaction on the target."""
session, sent = _Session("😂"), []
target = _channel("Alice: hi all")
screen = _screen([target], session, sent)
screen.handle("up")
assert "^E react" in screen.footer_hint
screen.handle("react")
await asyncio.sleep(0)
await asyncio.gather(*screen._tasks)
assert session.asked == ["React — Alice"]
assert sent == [reactions.channel_text("😂", "Alice", "hi all", _T)]
assert screen._messages == [target]
assert screen._fold.reacted(target, "😂")
def test_a_room_shows_reaction_text_as_it_arrived() -> None:
"""A room has no reactions. Its messages stay messages, and ^E offers nothing."""
body = reactions.direct_text("👍", "x", _T)
room = Contact(name="Board", public_key="ab" * 32, key_prefix="ab" * 6, node_type=3)
screen = ChatScreen(
Conversation(label="Board", is_channel=False, contact=room),
[ChatMessage(text=body, peer="abababababab", created_at=_at(_T))],
send=None,
names={},
session=_Session(None),
)
assert len(screen._messages) == 1
screen.handle("up")
assert not screen._can_react()
def test_the_messages_list_previews_a_reaction_in_words() -> None:
"""The latest message of a conversation can be a reaction. Its preview says what it is."""
from meshterm.tools.chat import _preview_text
react = "Bob: " + reactions.channel_text("👍", "Alice", "hi", _T)
preview = _preview_text(_channel(react), lambda name: None).plain
assert "Bob" in preview and "reacted 👍" in preview and "@[" not in preview
ours = ChatMessage(text=reactions.direct_text("😂", "hi", _T), outbound=True)
assert "reacted 😂" in _preview_text(ours, lambda name: None).plain
def test_a_reaction_from_a_node_with_a_long_name_folds() -> None:
"""A node name can have 31 characters. A sender of 22 is still split off its reaction.
The two messages are from the hardware log that found the fault: the name limit was 20,
so the reaction of ``Hazelwood r0b3.Anduril`` showed as a new message.
"""
fold = ReactionFold(channel=True)
target = _channel("Hazelwood: ⚽ Union 2-1 at Chicago Fire (59')", 1_792_029_645)
fold.take(target)
react = _channel("Hazelwood r0b3.Anduril: @[Hazelwood]🔥\nmzjsjjpk", 1_792_029_654)
assert fold.take(react) is True
(badge,) = fold.badges(target)
assert badge.emoji == "🔥" and badge.reactors == [reactions.Reactor("Hazelwood r0b3.Anduril")]
async def test_the_reaction_dialog_searches_every_emoji_by_name() -> None:
"""The preferred emoji come first, then each other emoji. A typed name filters them all."""
from meshterm.ui.chat import emoji_catalog
from meshterm.ui.tui import Choice, Separator
session = _Session(None)
screen = _screen([_channel("Alice: hi all")], session, [])
screen.handle("up")
screen.handle("react")
await asyncio.sleep(0)
await asyncio.gather(*screen._tasks)
assert session.filterable is True
values = [item.value for item in session.items if isinstance(item, Choice)]
assert values[:6] == reactions.DEFAULT_REACTIONS.split()
assert "🔥" in values and values.count("👍") == 1 # a preferred emoji is listed once
assert isinstance(session.items[0], Separator) # "Your reactions", then "All emoji"
assert sum(isinstance(item, Separator) for item in session.items) == 2
fire = next(i for i in session.items if isinstance(i, Choice) and i.value == "🔥")
assert "fire" in fire.label.plain
assert len(emoji_catalog()) > 1000
assert all(reactions.is_reaction_emoji(emoji) for emoji in emoji_catalog())
+9 -7
View File
@@ -71,12 +71,14 @@ class _TracingDevice(MockDevice):
await super().set_flood_scope(region)
self.trail.append(("scope", key))
async def send_channel_message(self, index: int, text: str) -> None:
await super().send_channel_message(index, text)
async def send_channel_message(
self, index: int, text: str, *, timestamp: int | None = None
) -> None:
await super().send_channel_message(index, text, timestamp=timestamp)
self.trail.append(("channel", text, self._send_scope))
async def send_direct_message(self, contact, text): # noqa: ANN001, ANN201
ack = await super().send_direct_message(contact, text)
async def send_direct_message(self, contact, text, **sent): # noqa: ANN001, ANN003, ANN201
ack = await super().send_direct_message(contact, text, **sent)
self.trail.append(("dm", text, self._send_scope))
return ack
@@ -148,10 +150,10 @@ async def test_a_dm_cannot_interleave_into_the_window() -> None:
release = asyncio.Event()
original = device.send_channel_message
async def slow_send(index: int, text: str) -> None:
async def slow_send(index: int, text: str, **sent) -> None: # noqa: ANN003
entered.set()
await release.wait()
await original(index, text)
await original(index, text, **sent)
device.send_channel_message = slow_send # type: ignore[method-assign]
scoped = asyncio.ensure_future(device.send_channel_in_scope(0, "scoped", "lakeside"))
@@ -186,7 +188,7 @@ async def test_a_send_that_fails_still_restores_the_scope() -> None:
"""If the channel send raises an error, the session scope must not stay on its region."""
device = _TracingDevice()
async def boom(index: int, text: str) -> None:
async def boom(index: int, text: str, **sent) -> None: # noqa: ANN003
raise DeviceCommandError("rejected")
device.send_channel_message = boom # type: ignore[method-assign]