Jean-Pierre MartineauandClaude Opus 5.5
efc02ddced
A final pass over the docstrings and comments: "the own X of Y" is now "Y's own X" (84 places), a navigation entry is a screen on the stack, our node is our node, a paint is a paint, the quoted UX labels are in code font, and a list of punctuation glyphs is in code font. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
371 lines
14 KiB
Python
371 lines
14 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
"""Tests for MeshTerm's own BlueZ pairing (PIN pairing on Linux through a D-Bus agent).
|
|
|
|
The ``pair()`` of bleak for BlueZ never gives a PIN. BlueZ asks a pairing agent for the
|
|
PIN, and bleak registers no agent. Thus a companion that has a PIN could not bond on Linux
|
|
at all. :mod:`meshterm.core.bluez` registers an agent for the length of one pairing. These
|
|
tests run against a scripted substitute for the system bus. They use no BlueZ, no radio,
|
|
and no real address or PIN. ``dbus_fast`` is a Linux dependency of bleak. The tests need
|
|
only its ``Message`` type.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
|
|
import pytest
|
|
|
|
dbus_fast = pytest.importorskip("dbus_fast")
|
|
from dbus_fast import Message, MessageType # noqa: E402
|
|
|
|
from meshterm.core import bluez # noqa: E402
|
|
|
|
_ADDR = "00:11:22:33:44:55"
|
|
_DEV = "/org/bluez/hci0/dev_00_11_22_33_44_55"
|
|
_PIN = "123456"
|
|
|
|
|
|
def _agent_call(member: str, *body, path: str = "/agent") -> Message:
|
|
"""A method call that BlueZ sends to the agent."""
|
|
signature = {
|
|
"RequestPasskey": "o",
|
|
"RequestPinCode": "o",
|
|
"RequestConfirmation": "ou",
|
|
"RequestAuthorization": "o",
|
|
"AuthorizeService": "os",
|
|
"Cancel": "",
|
|
}[member]
|
|
return Message(
|
|
path=path,
|
|
interface="org.bluez.Agent1",
|
|
member=member,
|
|
signature=signature,
|
|
body=list(body),
|
|
serial=1,
|
|
)
|
|
|
|
|
|
def test_status_names_read_as_words() -> None:
|
|
"""A BlueZ error name becomes the words that the status sets of the connection use."""
|
|
assert bluez.status_name("org.bluez.Error.AuthenticationFailed") == "authentication failed"
|
|
assert bluez.status_name("org.bluez.Error.ConnectionAttemptFailed") == (
|
|
"connection attempt failed"
|
|
)
|
|
# The bare Failed gives no information, so the detail of BlueZ itself is added.
|
|
assert bluez.status_name("org.bluez.Error.Failed", ["le-connection-abort-by-local"]) == (
|
|
"failed: le-connection-abort-by-local"
|
|
)
|
|
|
|
|
|
def test_is_mac() -> None:
|
|
"""BlueZ can pair only a 48-bit address. It cannot pair a macOS UUID."""
|
|
assert bluez.is_mac(_ADDR) and bluez.is_mac("aa-bb-cc-dd-ee-ff")
|
|
assert not bluez.is_mac("550e8400-e29b-41d4-a716-446655440000")
|
|
assert not bluez.is_mac("")
|
|
|
|
|
|
def test_agent_answers_the_passkey_with_the_pin() -> None:
|
|
"""The agent answers RequestPasskey with the PIN as a number, and it notes that it was asked."""
|
|
agent = bluez.PinAgent("/agent", _PIN, _DEV)
|
|
reply = agent.handle(_agent_call("RequestPasskey", _DEV))
|
|
assert reply.message_type == MessageType.METHOD_RETURN
|
|
assert reply.signature == "u" and reply.body == [123456]
|
|
assert agent.asked
|
|
|
|
|
|
def test_agent_refuses_another_device_and_other_paths() -> None:
|
|
"""The agent never approves a pairing that MeshTerm did not start.
|
|
|
|
It ignores calls for other paths.
|
|
"""
|
|
agent = bluez.PinAgent("/agent", _PIN, _DEV)
|
|
reply = agent.handle(_agent_call("RequestPasskey", "/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF"))
|
|
assert reply.message_type == MessageType.ERROR
|
|
assert reply.error_name == "org.bluez.Error.Rejected"
|
|
assert not agent.asked
|
|
# The call is not for the path of the agent, so the agent ignores it.
|
|
assert agent.handle(_agent_call("RequestPasskey", _DEV, path="/elsewhere")) is None
|
|
|
|
|
|
def test_agent_confirms_only_the_number_its_pin_names() -> None:
|
|
"""The agent approves Numeric Comparison only for the number of our own PIN."""
|
|
agent = bluez.PinAgent("/agent", _PIN, _DEV)
|
|
ok = agent.handle(_agent_call("RequestConfirmation", _DEV, 123456))
|
|
assert ok.message_type == MessageType.METHOD_RETURN
|
|
wrong = agent.handle(_agent_call("RequestConfirmation", _DEV, 654321))
|
|
assert wrong.message_type == MessageType.ERROR
|
|
|
|
|
|
class _Reply:
|
|
"""A reply as ``bus.call`` returns it."""
|
|
|
|
def __init__(self, body=None, error: str | None = None, detail: str = "") -> None:
|
|
self.message_type = MessageType.ERROR if error else MessageType.METHOD_RETURN
|
|
self.error_name = error
|
|
self.body = [detail] if error else (body or [])
|
|
|
|
|
|
class _Variant:
|
|
def __init__(self, value) -> None:
|
|
self.value = value
|
|
|
|
|
|
class _FakeBus:
|
|
"""A scripted system bus: the object tree of BlueZ and the answer to ``Pair``.
|
|
|
|
``pair_reply`` is what ``Device1.Pair`` returns. For a list, the bus gives one entry
|
|
for each call. This scripts a retry. ``asks`` makes BlueZ call ``RequestPasskey`` of the
|
|
agent first, as a real Passkey Entry pairing does.
|
|
"""
|
|
|
|
def __init__(self, *, known=True, paired=False, pair_reply=None, asks=True) -> None:
|
|
self.known, self.paired = known, paired
|
|
self.pair_reply = pair_reply or _Reply()
|
|
self.asks = asks
|
|
self.calls: list[str] = []
|
|
self.handlers: list = []
|
|
self.agent_answer = None
|
|
self.disconnected = False
|
|
|
|
def _objects(self) -> dict:
|
|
objects = {"/org/bluez/hci0": {"org.bluez.Adapter1": {}}}
|
|
if self.known:
|
|
objects[_DEV] = {
|
|
"org.bluez.Device1": {
|
|
"Address": _Variant(_ADDR),
|
|
"Paired": _Variant(self.paired),
|
|
"Adapter": _Variant("/org/bluez/hci0"),
|
|
}
|
|
}
|
|
return objects
|
|
|
|
async def call(self, msg):
|
|
self.calls.append(msg.member)
|
|
if msg.member == "GetManagedObjects":
|
|
return _Reply([self._objects()])
|
|
if msg.member == "RemoveDevice":
|
|
self.known = self.paired = False
|
|
return _Reply()
|
|
if msg.member == "StartDiscovery":
|
|
self.known = True # the adapter hears the device again
|
|
return _Reply()
|
|
if msg.member == "Pair":
|
|
if self.asks:
|
|
for handler in self.handlers:
|
|
self.agent_answer = handler(_agent_call("RequestPasskey", _DEV, path=_AGENT))
|
|
if isinstance(self.pair_reply, list):
|
|
return self.pair_reply.pop(0)
|
|
return self.pair_reply
|
|
return _Reply()
|
|
|
|
def add_message_handler(self, handler) -> None:
|
|
self.handlers.append(handler)
|
|
|
|
def remove_message_handler(self, handler) -> None:
|
|
self.handlers.remove(handler)
|
|
|
|
def disconnect(self) -> None:
|
|
self.disconnected = True
|
|
|
|
|
|
_AGENT = f"/net/meshterm/agent{os.getpid()}"
|
|
|
|
|
|
@pytest.fixture
|
|
def bus(monkeypatch):
|
|
"""Point :mod:`bluez` at a new fake bus."""
|
|
holder: dict = {}
|
|
|
|
async def _system_bus():
|
|
return holder["bus"]
|
|
|
|
monkeypatch.setattr(bluez, "_system_bus", _system_bus)
|
|
|
|
def _make(**kwargs) -> _FakeBus:
|
|
holder["bus"] = _FakeBus(**kwargs)
|
|
return holder["bus"]
|
|
|
|
return _make
|
|
|
|
|
|
def test_pair_registers_an_agent_answers_the_pin_and_trusts(bus) -> None:
|
|
"""The normal path of a pairing.
|
|
|
|
The code registers the agent, calls Pair, answers the PIN, sets Trusted, unregisters
|
|
the agent, and closes the bus.
|
|
"""
|
|
fake = bus()
|
|
assert asyncio.run(bluez.pair(_ADDR, _PIN, force=False, discover_s=0)) == ("paired", "")
|
|
# Disconnect: Pair leaves its link up, and a companion that is connected stops advertising.
|
|
assert fake.calls[1:] == ["RegisterAgent", "Pair", "Set", "Disconnect", "UnregisterAgent"]
|
|
assert fake.agent_answer.body == [123456] # the PIN got to BlueZ as the passkey
|
|
assert fake.handlers == [] and fake.disconnected
|
|
|
|
|
|
def test_pair_trusts_an_existing_bond_unless_forced(bus) -> None:
|
|
"""The code uses an existing bond again. ``force`` removes the bond and pairs again."""
|
|
fake = bus(paired=True)
|
|
assert asyncio.run(bluez.pair(_ADDR, _PIN, force=False, discover_s=0)) == ("reused", "")
|
|
assert "Pair" not in fake.calls
|
|
|
|
fake = bus(paired=True)
|
|
assert asyncio.run(bluez.pair(_ADDR, _PIN, force=True, discover_s=1)) == ("paired", "")
|
|
assert fake.calls.index("RemoveDevice") < fake.calls.index("Pair")
|
|
|
|
|
|
def test_pair_reports_a_wrong_pin_and_a_refusal_apart(bus) -> None:
|
|
"""The code tells a wrong PIN and a refusal apart.
|
|
|
|
AuthenticationFailed after the agent was asked for the PIN is a wrong PIN. Before the
|
|
agent was asked, it is a refusal.
|
|
"""
|
|
bus(pair_reply=_Reply(error="org.bluez.Error.AuthenticationFailed"))
|
|
assert asyncio.run(bluez.pair(_ADDR, _PIN, force=False, discover_s=0)) == (
|
|
"failed",
|
|
"authentication failed",
|
|
)
|
|
bus(pair_reply=_Reply(error="org.bluez.Error.AuthenticationFailed"), asks=False)
|
|
assert asyncio.run(bluez.pair(_ADDR, _PIN, force=False, discover_s=0)) == (
|
|
"failed",
|
|
"authentication rejected",
|
|
)
|
|
|
|
|
|
def test_pair_listens_for_a_forgotten_device_then_gives_up(bus) -> None:
|
|
"""The code listens for a device that BlueZ forgot.
|
|
|
|
A device that it never hears is ``absent``.
|
|
"""
|
|
fake = bus(known=False)
|
|
assert asyncio.run(bluez.pair(_ADDR, _PIN, force=False, discover_s=1))[0] == "paired"
|
|
assert fake.calls.count("StartDiscovery") == fake.calls.count("StopDiscovery") == 1
|
|
|
|
|
|
def test_pair_without_a_bus_is_an_error_not_a_raise(monkeypatch) -> None:
|
|
"""If there is no system bus (a container, no BlueZ), the code reports ``error``.
|
|
|
|
It does not raise an exception.
|
|
"""
|
|
|
|
async def _no_bus():
|
|
raise OSError("no system bus")
|
|
|
|
monkeypatch.setattr(bluez, "_system_bus", _no_bus)
|
|
outcome, status = asyncio.run(bluez.pair(_ADDR, _PIN, force=False))
|
|
assert outcome == "error" and "no system bus" in status
|
|
|
|
|
|
def test_a_pin_that_is_not_digits_never_reaches_bluez(bus) -> None:
|
|
"""A PIN that is not digits never reaches BlueZ.
|
|
|
|
A passkey is a number. Any other value is a wrong PIN, and the code does not ask the
|
|
radio.
|
|
"""
|
|
fake = bus()
|
|
assert asyncio.run(bluez.pair(_ADDR, "12ab56", force=False))[0] == "failed"
|
|
assert fake.calls == []
|
|
|
|
|
|
def test_pair_retries_a_link_that_failed_to_come_up(bus, monkeypatch) -> None:
|
|
"""The code retries a link that failed to come up.
|
|
|
|
ConnectionAttemptFailed is a fault of the radio and not of the pairing. The code calls
|
|
Pair again, a limited number of times. bleak retries these faults inside its own
|
|
connect. The Pair of BlueZ does not. On a uConsole, the first Pair failed in this way.
|
|
Then the connect went on with no pairing, and it stopped for 32 s.
|
|
"""
|
|
monkeypatch.setattr(bluez, "PAIR_RETRY_DELAY_S", 0)
|
|
flake = _Reply(error="org.bluez.Error.ConnectionAttemptFailed")
|
|
fake = bus(pair_reply=[flake, flake, _Reply()])
|
|
assert asyncio.run(bluez.pair(_ADDR, _PIN, force=False, discover_s=0)) == ("paired", "")
|
|
assert fake.calls.count("Pair") == 3
|
|
|
|
fake = bus(pair_reply=[flake] * bluez.PAIR_ATTEMPTS)
|
|
outcome, status = asyncio.run(bluez.pair(_ADDR, _PIN, force=False, discover_s=0))
|
|
assert (outcome, status) == ("failed", "connection attempt failed")
|
|
assert fake.calls.count("Pair") == bluez.PAIR_ATTEMPTS # the number of retries has a limit
|
|
|
|
|
|
def test_pair_never_retries_a_wrong_pin(bus, monkeypatch) -> None:
|
|
"""The code never retries a wrong PIN.
|
|
|
|
It retries only a link failure. A passkey that BlueZ refuses is an answer.
|
|
"""
|
|
monkeypatch.setattr(bluez, "PAIR_RETRY_DELAY_S", 0)
|
|
fake = bus(pair_reply=[_Reply(error="org.bluez.Error.AuthenticationFailed"), _Reply()])
|
|
assert asyncio.run(bluez.pair(_ADDR, _PIN, force=False, discover_s=0))[0] == "failed"
|
|
assert fake.calls.count("Pair") == 1
|
|
|
|
|
|
def test_agent_without_a_pin_refuses_at_once() -> None:
|
|
"""An agent with no PIN refuses at once.
|
|
|
|
If there is no PIN, the agent refuses each passkey question immediately. It does not
|
|
leave the question without an answer.
|
|
"""
|
|
agent = bluez.PinAgent("/agent", None, bluez.device_tail(_ADDR))
|
|
for member, body in (("RequestPasskey", [_DEV]), ("RequestConfirmation", [_DEV, 0])):
|
|
reply = agent.handle(_agent_call(member, *body))
|
|
assert reply.message_type == MessageType.ERROR, member
|
|
assert reply.error_name == "org.bluez.Error.Rejected"
|
|
|
|
|
|
def test_agent_never_approves_a_pairing_without_the_passkey() -> None:
|
|
"""The agent never approves a pairing without the passkey.
|
|
|
|
The agent refuses a "Just Works" yes or no, also when it has a PIN. This is how a bond
|
|
becomes weak.
|
|
"""
|
|
agent = bluez.PinAgent("/agent", _PIN, _DEV)
|
|
reply = agent.handle(_agent_call("RequestAuthorization", _DEV))
|
|
assert reply.message_type == MessageType.ERROR
|
|
|
|
|
|
def test_agent_matches_a_device_by_its_path_tail() -> None:
|
|
"""The agent matches a device by the tail of its path.
|
|
|
|
Before a connect, the adapter is not known. Thus the agent matches the ``/dev_…`` tail.
|
|
"""
|
|
agent = bluez.PinAgent("/agent", _PIN, bluez.device_tail("00-11-22-33-44-55"))
|
|
assert agent.handle(_agent_call("RequestPasskey", _DEV)).body == [123456]
|
|
reply = agent.handle(_agent_call("RequestPasskey", "/org/bluez/hci1/dev_AA_BB_CC_DD_EE_FF"))
|
|
assert reply.message_type == MessageType.ERROR
|
|
|
|
|
|
def test_answering_is_the_default_agent_for_the_connect_and_then_steps_down(bus) -> None:
|
|
"""The agent is the default agent for the connect, and then it steps down.
|
|
|
|
The order is: the code registers the agent, makes it the default, then unregisters it
|
|
and closes the bus.
|
|
"""
|
|
fake = bus()
|
|
|
|
async def scenario() -> list[str]:
|
|
async with bluez.answering(_ADDR, None):
|
|
during = list(fake.calls)
|
|
return during
|
|
|
|
during = asyncio.run(scenario())
|
|
assert during == ["RegisterAgent", "RequestDefaultAgent"]
|
|
assert fake.calls[-1] == "UnregisterAgent"
|
|
assert fake.handlers == [] and fake.disconnected
|
|
|
|
|
|
def test_answering_without_a_bus_still_runs_the_connect(monkeypatch) -> None:
|
|
"""If there is no system bus, the connect still runs.
|
|
|
|
It has no agent, and nothing raises an exception.
|
|
"""
|
|
|
|
async def _no_bus():
|
|
raise OSError("no system bus")
|
|
|
|
monkeypatch.setattr(bluez, "_system_bus", _no_bus)
|
|
|
|
async def scenario() -> str:
|
|
async with bluez.answering(_ADDR, _PIN):
|
|
return "connected"
|
|
|
|
assert asyncio.run(scenario()) == "connected"
|