fix: verify the device's mutual-auth server proof (M8)

src/opendisplay/crypto.py

@@ -77,6 +77,21 @@ def compute_challenge_response(
     return aes_cmac(master_key, server_nonce + client_nonce + device_id)
 
 
+def compute_server_proof(
+    session_key: bytes,
+    server_nonce: bytes,
+    client_nonce: bytes,
+    device_id: bytes = _DEVICE_ID,
+) -> bytes:
+    """Compute the device's mutual-auth proof returned in step 2 of auth.
+
+    Matches firmware: CMAC(session_key, server_nonce || client_nonce || device_id).
+    The client recomputes this to authenticate the device — a peer that returns
+    status OK without knowing the master key cannot produce it.
+    """
+    return aes_cmac(session_key, server_nonce + client_nonce + device_id)
+
+
 def get_nonce(session_id: bytes, counter: int) -> bytes:
     """Build the 16-byte full nonce: session_id(8) || counter_be(8)."""
     return session_id + counter.to_bytes(8, "big")

src/opendisplay/device.py

@@ -5,6 +5,7 @@
 
 import asyncio
 import functools
+import hmac
 import logging
 import time
 from collections.abc import AsyncIterator, Awaitable, Callable
@@ -16,6 +17,7 @@
 
 from .crypto import (
     compute_challenge_response,
+    compute_server_proof,
     decrypt_response,
     derive_session_id,
     derive_session_key,
@@ -34,6 +36,7 @@
     zlib_window_bits,
 )
 from .exceptions import (
+    AuthenticationFailedError,
     AuthenticationRequiredError,
     AuthenticationSessionExistsError,
     BLETimeoutError,
@@ -591,10 +594,20 @@ async def authenticate(self, key: bytes) -> None:
         challenge = compute_challenge_response(key, server_nonce, client_nonce, device_id)
         await self._conn.write_command(build_authenticate_step2(client_nonce, challenge))
         success_response = await self._conn.read_response(timeout=self.TIMEOUT_ACK)
-        parse_authenticate_success(success_response)  # raises on wrong key / error
+        server_proof = parse_authenticate_success(success_response)  # raises on wrong key / error
 
         # Derive session key and ID
-        self._session_key = derive_session_key(key, client_nonce, server_nonce, device_id)
+        session_key = derive_session_key(key, client_nonce, server_nonce, device_id)
+
+        # Verify the device's mutual-auth proof so we authenticate the device, not
+        # just the other way around. A device (or MITM) that returns status OK
+        # without knowing the master key cannot produce this CMAC. Constant-time
+        # compare to avoid leaking a timing side channel.
+        expected_proof = compute_server_proof(session_key, server_nonce, client_nonce, device_id)
+        if not hmac.compare_digest(server_proof, expected_proof):
+            raise AuthenticationFailedError("Device failed mutual authentication (server proof mismatch)")
+
+        self._session_key = session_key
         self._session_id = derive_session_id(self._session_key, client_nonce, server_nonce)
         self._nonce_counter = 0
         self._auth_time = time.monotonic()

tests/unit/test_auth_server_proof.py

@@ -0,0 +1,63 @@
+"""Tests for mutual authentication — verifying the device's server proof (M8)."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from epaper_dithering import ColorScheme
+
+from opendisplay import OpenDisplayDevice
+from opendisplay.crypto import _DEVICE_ID, compute_server_proof, derive_session_key
+from opendisplay.exceptions import AuthenticationFailedError
+from opendisplay.models.capabilities import DeviceCapabilities
+
+_KEY = bytes(range(16))
+_SERVER_NONCE = bytes(range(100, 116))
+_CLIENT_NONCE = bytes(range(200, 216))
+
+
+class _FakeConn:
+    def __init__(self, responses: list[bytes]) -> None:
+        self._responses = responses
+
+    async def write_command(self, data: bytes) -> None:
+        pass
+
+    async def read_response(self, timeout: float) -> bytes:
+        return self._responses.pop(0)
+
+
+def _device(success_proof: bytes) -> OpenDisplayDevice:
+    dev = OpenDisplayDevice(
+        mac_address="AA:BB:CC:DD:EE:FF",
+        capabilities=DeviceCapabilities(width=2, height=2, color_scheme=ColorScheme.MONO),
+    )
+    challenge = b"\x00\x50\x00" + _SERVER_NONCE  # old format -> default device_id
+    success = b"\x00\x50\x00" + success_proof  # status OK + 16-byte proof
+    dev._connection = _FakeConn([challenge, success])  # type: ignore[assignment]
+    return dev
+
+
+def _good_proof() -> bytes:
+    session_key = derive_session_key(_KEY, _CLIENT_NONCE, _SERVER_NONCE, _DEVICE_ID)
+    return compute_server_proof(session_key, _SERVER_NONCE, _CLIENT_NONCE, _DEVICE_ID)
+
+
+def test_authenticate_accepts_valid_server_proof(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr("opendisplay.device.generate_client_nonce", lambda: _CLIENT_NONCE)
+    device = _device(_good_proof())
+
+    asyncio.run(device.authenticate(_KEY))
+
+    assert device._session_key is not None
+
+
+def test_authenticate_rejects_wrong_server_proof(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr("opendisplay.device.generate_client_nonce", lambda: _CLIENT_NONCE)
+    device = _device(b"\xff" * 16)  # bogus proof a MITM would send
+
+    with pytest.raises(AuthenticationFailedError, match="mutual authentication"):
+        asyncio.run(device.authenticate(_KEY))
+
+    assert device._session_key is None