Danil Iarosh
add8aff0af
- Добавлена обратная совместимость с прошивками, содержащими маркер no_auth. - Повторная авторизация после команды 04 и успешного обновления прошивки приведена к требуемому порядку. - В терминал команд добавлена ручная кнопка авторизации. - Реализовано OTA-обновление приложения с загрузкой платформенного артефакта, безопасной заменой и перезапуском. - Проверка обновлений учитывает только более новые версии по схеме major, minor и patch. - Обновлены имена Windows и Linux сборок, документация и тестовое покрытие.
263 lines
8.9 KiB
Python
263 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from collections import deque
|
|
from unittest.mock import patch
|
|
|
|
from app.services.auth_service import (
|
|
AuthStatus,
|
|
_aes_encrypt,
|
|
check_status,
|
|
firmware_requires_auth,
|
|
perform_handshake,
|
|
perform_required_handshake,
|
|
perform_startup_handshake,
|
|
wait_for_startup_banner,
|
|
)
|
|
|
|
|
|
class FakeConnection:
|
|
def __init__(self, responses: list[str | None]) -> None:
|
|
self.is_connected = True
|
|
self._responses = deque(responses)
|
|
self.events: list[str] = []
|
|
self.release_count = 0
|
|
self.firmware_version: str | None = None
|
|
self.auth_required: bool | None = None
|
|
|
|
def read_response(self, timeout: float | None = None) -> str | None:
|
|
response = self._responses.popleft() if self._responses else None
|
|
self.events.append(f"read:{response}")
|
|
return response
|
|
|
|
def send_command(self, command: str) -> None:
|
|
self.events.append(f"send:{command}")
|
|
|
|
def write_raw(self, command: str) -> None:
|
|
self.events.append(f"write:{command}")
|
|
|
|
def release(self) -> None:
|
|
self.release_count += 1
|
|
|
|
def set_firmware_version(self, version: str, *, auth_required: bool) -> None:
|
|
self.firmware_version = version
|
|
self.auth_required = auth_required
|
|
|
|
|
|
class StartupHandshakeTests(unittest.TestCase):
|
|
_HOST_CHALLENGE = bytes.fromhex("00112233445566778899AABBCCDDEEFF")
|
|
_CHAL = "CHAL00112233445566778899AABBCCDDEEFF"
|
|
_RAES = "RAES40E9A18C22C2CFCD764666C90B358F9D"
|
|
|
|
def _perform_handshake(
|
|
self,
|
|
conn: FakeConnection,
|
|
*,
|
|
timeout: float = 0.1,
|
|
ready_timeout: float = 0.1,
|
|
):
|
|
with patch(
|
|
"app.services.auth_service.os.urandom",
|
|
return_value=self._HOST_CHALLENGE,
|
|
):
|
|
return perform_handshake(
|
|
conn,
|
|
timeout=timeout,
|
|
ready_timeout=ready_timeout,
|
|
)
|
|
|
|
def test_waits_for_banner_and_two_seconds_before_sending_auth(self) -> None:
|
|
conn = FakeConnection([
|
|
"boot log",
|
|
"periphery controller ver_2.3.6 release",
|
|
self._CHAL,
|
|
self._RAES,
|
|
"AUTH_PC_OK",
|
|
"READY!",
|
|
])
|
|
sleep_calls: list[float] = []
|
|
|
|
def fake_sleep(seconds: float) -> None:
|
|
sleep_calls.append(seconds)
|
|
conn.events.append(f"sleep:{seconds}")
|
|
|
|
with patch(
|
|
"app.services.auth_service.os.urandom",
|
|
return_value=self._HOST_CHALLENGE,
|
|
):
|
|
result = perform_startup_handshake(
|
|
conn,
|
|
startup_delay=2.0,
|
|
sleep_fn=fake_sleep,
|
|
)
|
|
|
|
self.assertEqual(AuthStatus.OK, result.status)
|
|
self.assertTrue(result.startup_ready)
|
|
self.assertEqual([2.0], sleep_calls)
|
|
sleep_index = conn.events.index("sleep:2.0")
|
|
send_index = next(
|
|
index for index, event in enumerate(conn.events) if event.startswith("send:07")
|
|
)
|
|
self.assertLess(sleep_index, send_index)
|
|
self.assertEqual("2.3.6 release", conn.firmware_version)
|
|
self.assertTrue(conn.auth_required)
|
|
|
|
def test_no_auth_banner_skips_handshake(self) -> None:
|
|
conn = FakeConnection([
|
|
"boot log",
|
|
"periphery controller ver_2.3.6_No_Auth_prizes",
|
|
])
|
|
sleep_calls: list[float] = []
|
|
|
|
result = perform_startup_handshake(
|
|
conn,
|
|
startup_delay=2.0,
|
|
sleep_fn=sleep_calls.append,
|
|
)
|
|
|
|
self.assertEqual(AuthStatus.OK, result.status)
|
|
self.assertTrue(result.startup_ready)
|
|
self.assertTrue(result.auth_skipped)
|
|
self.assertEqual("2.3.6_No_Auth_prizes", result.firmware_version)
|
|
self.assertFalse(conn.auth_required)
|
|
self.assertEqual([2.0], sleep_calls)
|
|
self.assertFalse(any(event.startswith("send:07") for event in conn.events))
|
|
self.assertEqual(1, conn.release_count)
|
|
|
|
def test_auth_mode_marker_is_case_insensitive(self) -> None:
|
|
self.assertFalse(firmware_requires_auth("2.3.6_no_auth_prizes"))
|
|
self.assertFalse(firmware_requires_auth("2.3.6_NO_AUTH"))
|
|
self.assertTrue(firmware_requires_auth("2.3.6_release"))
|
|
|
|
def test_manual_handshake_is_skipped_for_known_no_auth_firmware(self) -> None:
|
|
conn = FakeConnection([])
|
|
conn.set_firmware_version("2.3.6_no_auth", auth_required=False)
|
|
|
|
result = perform_required_handshake(conn, timeout=0.1)
|
|
|
|
self.assertEqual(AuthStatus.OK, result.status)
|
|
self.assertTrue(result.auth_skipped)
|
|
self.assertFalse(any(event.startswith("send:") for event in conn.events))
|
|
|
|
def test_banner_requires_non_empty_version_suffix(self) -> None:
|
|
conn = FakeConnection(["periphery controller ver_", None])
|
|
|
|
banner = wait_for_startup_banner(conn, timeout=0.1)
|
|
|
|
self.assertIsNone(banner)
|
|
self.assertFalse(any(event.startswith("send:") for event in conn.events))
|
|
|
|
def test_startup_timeout_never_sends_auth_command(self) -> None:
|
|
conn = FakeConnection(["unrelated boot log", None])
|
|
|
|
result = perform_startup_handshake(
|
|
conn,
|
|
banner_timeout=0.1,
|
|
startup_delay=2.0,
|
|
sleep_fn=lambda _seconds: self.fail("Пауза не должна начинаться без banner"),
|
|
)
|
|
|
|
self.assertEqual(AuthStatus.TIMEOUT, result.status)
|
|
self.assertFalse(result.startup_ready)
|
|
self.assertFalse(any(event.startswith("send:") for event in conn.events))
|
|
|
|
def test_raes_does_not_complete_handshake(self) -> None:
|
|
conn = FakeConnection([self._CHAL, self._RAES, None])
|
|
|
|
result = self._perform_handshake(conn)
|
|
|
|
self.assertEqual(AuthStatus.TIMEOUT, result.status)
|
|
self.assertIn("READY!", result.detail)
|
|
|
|
def test_auth_pc_ok_does_not_complete_handshake(self) -> None:
|
|
conn = FakeConnection([self._CHAL, self._RAES, "AUTH_PC_OK", None])
|
|
|
|
result = self._perform_handshake(conn)
|
|
|
|
self.assertEqual(AuthStatus.TIMEOUT, result.status)
|
|
self.assertIn("READY!", result.detail)
|
|
|
|
def test_ready_without_exclamation_mark_is_rejected(self) -> None:
|
|
conn = FakeConnection([self._CHAL, self._RAES, "READY", None])
|
|
|
|
result = self._perform_handshake(conn)
|
|
|
|
self.assertEqual(AuthStatus.TIMEOUT, result.status)
|
|
|
|
def test_exact_ready_bang_completes_handshake(self) -> None:
|
|
conn = FakeConnection([
|
|
self._CHAL,
|
|
self._RAES,
|
|
"AUTH_PC_OK",
|
|
" READY! ",
|
|
])
|
|
|
|
result = self._perform_handshake(conn)
|
|
|
|
self.assertEqual(AuthStatus.OK, result.status)
|
|
self.assertEqual(1, conn.release_count)
|
|
|
|
def test_effective_aes_key_matches_controller_profile(self) -> None:
|
|
encrypted = _aes_encrypt(self._HOST_CHALLENGE)
|
|
|
|
self.assertEqual("40E9A18C22C2CFCD764666C90B358F9D", encrypted.hex().upper())
|
|
|
|
def test_invalid_raes_fails_controller_verification(self) -> None:
|
|
conn = FakeConnection([
|
|
self._CHAL,
|
|
"RAES00000000000000000000000000000000",
|
|
"READY!",
|
|
])
|
|
|
|
result = self._perform_handshake(conn)
|
|
|
|
self.assertEqual(AuthStatus.FAILED, result.status)
|
|
self.assertIn("RAES", result.detail)
|
|
|
|
def test_status_requires_exact_ready_bang(self) -> None:
|
|
ready_conn = FakeConnection(["READY!"])
|
|
not_ready_conn = FakeConnection(["READY", None])
|
|
|
|
self.assertEqual(AuthStatus.OK, check_status(ready_conn, timeout=0.1))
|
|
self.assertEqual(AuthStatus.NEED_AUTH, check_status(not_ready_conn, timeout=0.1))
|
|
|
|
def test_no_auth_status_accepts_any_non_empty_response(self) -> None:
|
|
conn = FakeConnection(["OK"])
|
|
conn.set_firmware_version("2.3.6_no_auth", auth_required=False)
|
|
|
|
self.assertEqual(AuthStatus.OK, check_status(conn, timeout=0.1))
|
|
|
|
def test_no_auth_status_timeout_is_connection_failure(self) -> None:
|
|
conn = FakeConnection([None])
|
|
conn.set_firmware_version("2.3.6_no_auth", auth_required=False)
|
|
|
|
self.assertEqual(AuthStatus.DISCONNECTED, check_status(conn, timeout=0.1))
|
|
|
|
def test_new_secure_banner_reenables_handshake(self) -> None:
|
|
conn = FakeConnection([
|
|
"periphery controller ver_2.0.0_release",
|
|
self._CHAL,
|
|
self._RAES,
|
|
"READY!",
|
|
])
|
|
conn.set_firmware_version("1.0.0_no_auth", auth_required=False)
|
|
|
|
with patch(
|
|
"app.services.auth_service.os.urandom",
|
|
return_value=self._HOST_CHALLENGE,
|
|
):
|
|
result = perform_startup_handshake(
|
|
conn,
|
|
startup_delay=0.0,
|
|
sleep_fn=lambda _seconds: None,
|
|
)
|
|
|
|
self.assertEqual(AuthStatus.OK, result.status)
|
|
self.assertFalse(result.auth_skipped)
|
|
self.assertTrue(conn.auth_required)
|
|
self.assertTrue(any(event.startswith("send:07") for event in conn.events))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|