diff --git a/src/opendisplay/crypto.py b/src/opendisplay/crypto.py index fbb89b5..1e8a3c4 100644 --- a/src/opendisplay/crypto.py +++ b/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") diff --git a/src/opendisplay/device.py b/src/opendisplay/device.py index f1415f3..974c1db 100644 --- a/src/opendisplay/device.py +++ b/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() diff --git a/tests/unit/test_auth_server_proof.py b/tests/unit/test_auth_server_proof.py new file mode 100644 index 0000000..10eefdc --- /dev/null +++ b/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