MeshTerm/tests/test_bluez.py

# 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"