From c34cb0cd7eb56c170cfffd79f90aa8c69967a8ec Mon Sep 17 00:00:00 2001 From: Danil Iarosh Date: Tue, 15 Sep 2026 10:49:51 +0300 Subject: [PATCH] feat: add authenticated firmware update flow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - добавлена полная повторная авторизация перед командой 04 - добавлена авторизация после успешного обновления и перезагрузки платы - общий OTA-сценарий подключён к базовому и PRO-интерфейсам - в терминал команд добавлена кнопка ручной авторизации - состояние авторизации синхронизировано с основной вкладкой - добавлены тесты порядка авторизации и обработки ошибок OTA - версия приложения повышена до 1.5.0 - обновлена документация --- README.md | 3 + app/__init__.py | 2 +- app/services/firmware_service.py | 119 ++++++++++++++++++ app/widgets/connection_tab.py | 27 +++- .../dialogs/command_terminal_dialog.py | 89 +++++++++++-- app/widgets/dialogs/firmware_dialog.py | 63 ++-------- app/widgets/sections/firmware_section.py | 39 ++---- tests/test_firmware_service.py | 102 +++++++++++++++ 8 files changed, 354 insertions(+), 90 deletions(-) create mode 100644 app/services/firmware_service.py create mode 100644 tests/test_firmware_service.py diff --git a/README.md b/README.md index 88ad296..3f7cc78 100644 --- a/README.md +++ b/README.md @@ -118,6 +118,9 @@ - Зеленый — подключено, плата готова к работе. Кнопка переподключения (↻) позволяет быстро восстановить связь при обрыве. +Перед запуском OTA и после успешного обновления приложение автоматически +выполняет полную AES-256 авторизацию платы. В терминале команд доступна +отдельная кнопка ручной авторизации. --- diff --git a/app/__init__.py b/app/__init__.py index 886cfa5..918dee7 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -1,3 +1,3 @@ """Plate Config App package.""" -__version__ = "1.4.2" +__version__ = "1.5.0" diff --git a/app/services/firmware_service.py b/app/services/firmware_service.py new file mode 100644 index 0000000..c352af1 --- /dev/null +++ b/app/services/firmware_service.py @@ -0,0 +1,119 @@ +"""Авторизованный запуск OTA-обновления контроллера.""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from dataclasses import dataclass + +from .auth_service import ( + AuthResult, + AuthStatus, + READY_TIMEOUT_SECONDS, + STARTUP_BANNER_TIMEOUT_SECONDS, + STARTUP_DELAY_SECONDS, + perform_handshake, + perform_startup_handshake, +) +from .com_service import ComConnection + +OTA_TIMEOUT_SECONDS = 60.0 + + +@dataclass +class FirmwareUpdateResult: + success: bool + detail: str + pre_auth: AuthResult | None = None + post_auth: AuthResult | None = None + + +def perform_authenticated_ota( + conn: ComConnection, + *, + log_message: Callable[[str], None] | None = None, + ota_timeout: float = OTA_TIMEOUT_SECONDS, + banner_timeout: float = STARTUP_BANNER_TIMEOUT_SECONDS, + startup_delay: float = STARTUP_DELAY_SECONDS, + handshake_timeout: float = 2.0, + ready_timeout: float = READY_TIMEOUT_SECONDS, + sleep_fn: Callable[[float], None] = time.sleep, +) -> FirmwareUpdateResult: + """Повторно авторизоваться, выполнить ``04`` и авторизоваться после OTA.""" + emit = log_message or (lambda _message: None) + if not conn.is_connected: + return FirmwareUpdateResult(False, "Порт не подключён") + + emit("Авторизация перед запуском OTA…") + try: + pre_auth = perform_handshake( + conn, + timeout=handshake_timeout, + ready_timeout=ready_timeout, + ) + except Exception as exc: + return FirmwareUpdateResult(False, f"Ошибка авторизации перед OTA: {exc}") + + if pre_auth.status != AuthStatus.OK: + return FirmwareUpdateResult( + False, + f"Авторизация перед OTA не выполнена: {pre_auth.detail}", + pre_auth=pre_auth, + ) + emit("✓ Авторизация перед OTA выполнена.") + + try: + conn.send_command("04") + emit("→ 04") + + deadline = time.monotonic() + ota_timeout + update_confirmed = False + while time.monotonic() < deadline and conn.is_connected: + remaining = max(deadline - time.monotonic(), 0.1) + response = conn.read_response(timeout=min(remaining, 2.0)) + if response is None: + continue + emit(f"← {response}") + if "successful update" in response.lower(): + update_confirmed = True + break + + if not update_confirmed: + return FirmwareUpdateResult( + False, + "Таймаут ожидания подтверждения успешного обновления", + pre_auth=pre_auth, + ) + + emit("✓ Обновление подтверждено. Ожидание перезагрузки и авторизации…") + post_auth = perform_startup_handshake( + conn, + banner_timeout=banner_timeout, + startup_delay=startup_delay, + handshake_timeout=handshake_timeout, + ready_timeout=ready_timeout, + sleep_fn=sleep_fn, + ) + if post_auth.status != AuthStatus.OK: + return FirmwareUpdateResult( + False, + f"Авторизация после OTA не выполнена: {post_auth.detail}", + pre_auth=pre_auth, + post_auth=post_auth, + ) + + emit("✓ Авторизация после OTA выполнена.") + return FirmwareUpdateResult( + True, + "Прошивка обновлена, плата перезагружена и авторизована", + pre_auth=pre_auth, + post_auth=post_auth, + ) + except Exception as exc: + return FirmwareUpdateResult( + False, + f"Ошибка OTA: {exc}", + pre_auth=pre_auth, + ) + finally: + conn.release() diff --git a/app/widgets/connection_tab.py b/app/widgets/connection_tab.py index 3570f9f..49001e3 100644 --- a/app/widgets/connection_tab.py +++ b/app/widgets/connection_tab.py @@ -630,11 +630,36 @@ class ConnectionTab(QWidget): def _open_terminal(self) -> None: from .dialogs.command_terminal_dialog import CommandTerminalDialog + + def create_terminal() -> CommandTerminalDialog: + dialog = CommandTerminalDialog(self._connection, parent=self) + dialog.authorization_finished.connect(self._on_terminal_authorization) + return dialog + self._exec_dialog( "terminal", - lambda: CommandTerminalDialog(self._connection, parent=self), + create_terminal, ) + def _on_terminal_authorization(self, result: AuthResult) -> None: + if result.status == AuthStatus.OK: + self._update_plate_state(PlateState.READY) + self._set_board_actions_enabled(True) + self._health_fail_count = 0 + self._apply_health_settings() + if not self._pro_toggle.isChecked(): + self._start_basic_polling() + return + + state = ( + PlateState.NEED_AUTH + if self._connection.is_connected + else PlateState.DISCONNECTED + ) + self._input_section.stop_polling() + self._set_board_actions_enabled(False) + self._update_plate_state(state) + def _not_implemented(self) -> None: QMessageBox.information(self, "В разработке", "Эта функция пока не реализована.") diff --git a/app/widgets/dialogs/command_terminal_dialog.py b/app/widgets/dialogs/command_terminal_dialog.py index 20a3b8c..d7ef585 100644 --- a/app/widgets/dialogs/command_terminal_dialog.py +++ b/app/widgets/dialogs/command_terminal_dialog.py @@ -5,7 +5,7 @@ from __future__ import annotations -from PySide6.QtCore import Qt +from PySide6.QtCore import QThread, Qt, Signal from PySide6.QtGui import QKeyEvent from PySide6.QtWidgets import ( QHBoxLayout, @@ -18,14 +18,19 @@ from PySide6.QtWidgets import ( ) from ...services.com_service import ComConnection -from ...ui.icons import get_icon_muted, icon +from ...services.auth_service import AuthResult, AuthStatus, perform_handshake +from ...ui.icons import get_icon_color, get_icon_muted, icon from .styled_dialog import StyledDialog class CommandTerminalDialog(StyledDialog): + authorization_finished = Signal(object) + def __init__(self, connection: ComConnection, parent=None) -> None: super().__init__(parent) self._conn = connection + self._auth_thread: _TerminalHandshakeThread | None = None + self._command_controls: list[QWidget] = [] self.setWindowTitle("Ввод команд") self.setMinimumSize(540, 400) @@ -74,11 +79,11 @@ class CommandTerminalDialog(StyledDialog): self._input.returnPressed.connect(self._send_command) input_layout.addWidget(self._input, 1) - send_btn = QPushButton("Отправить") - send_btn.setProperty("role", "primary") - send_btn.setIcon(icon("fa5s.paper-plane", color="#ecfdf5")) - send_btn.clicked.connect(self._send_command) - input_layout.addWidget(send_btn) + self._send_btn = QPushButton("Отправить") + self._send_btn.setProperty("role", "primary") + self._send_btn.setIcon(icon("fa5s.paper-plane", color="#ecfdf5")) + self._send_btn.clicked.connect(self._send_command) + input_layout.addWidget(self._send_btn) clear_btn = QPushButton("Очистить") clear_btn.setIcon(icon("fa5s.eraser", color=get_icon_muted())) @@ -89,6 +94,12 @@ class CommandTerminalDialog(StyledDialog): # ── Быстрые команды ── quick_layout = QHBoxLayout() + self._auth_btn = QPushButton("Авторизация") + self._auth_btn.setMinimumHeight(30) + self._auth_btn.setIcon(icon("fa5s.key", color=get_icon_color())) + self._auth_btn.clicked.connect(self._start_authorization) + quick_layout.addWidget(self._auth_btn) + quick_commands = [ ("Статус (99)", "99"), ("Версия (05)", "05"), @@ -102,6 +113,7 @@ class CommandTerminalDialog(StyledDialog): btn.setStyleSheet("padding: 4px 10px;") btn.clicked.connect(lambda checked=False, c=cmd: self._send(c)) quick_layout.addWidget(btn) + self._command_controls.append(btn) quick_layout.addStretch() bottom_layout.addLayout(quick_layout) @@ -121,6 +133,48 @@ class CommandTerminalDialog(StyledDialog): splitter.setStretchFactor(1, 25) root.addWidget(splitter, 1) + self._command_controls.extend((self._input, self._send_btn, self._auth_btn)) + + def _start_authorization(self) -> None: + if not self._conn.is_connected: + self._log.append('✗ Порт не подключён') + return + if self._auth_thread is not None: + return + + self._set_command_controls_enabled(False) + self._auth_btn.setText("Авторизация…") + self._log.append('Запуск полной авторизации…') + self._conn.pause_drain() + + thread = _TerminalHandshakeThread(self._conn, parent=self) + thread.result_ready.connect(self._on_authorization_done) + thread.finished.connect(self._cleanup_auth_thread) + self._auth_thread = thread + thread.start() + + def _on_authorization_done(self, result: AuthResult) -> None: + self._conn.resume_drain() + self._set_command_controls_enabled(True) + self._auth_btn.setText("Авторизация") + if result.status == AuthStatus.OK: + self._log.append( + '✓ Авторизация успешно завершена' + ) + else: + self._log.append( + f'✗ Авторизация: {result.detail}' + ) + self.authorization_finished.emit(result) + + def _cleanup_auth_thread(self) -> None: + if self._auth_thread is not None: + self._auth_thread.deleteLater() + self._auth_thread = None + + def _set_command_controls_enabled(self, enabled: bool) -> None: + for control in self._command_controls: + control.setEnabled(enabled) def _send_command(self) -> None: text = self._input.text().strip() @@ -186,8 +240,29 @@ class CommandTerminalDialog(StyledDialog): self._log.append(f'← {stripped}') def closeEvent(self, event) -> None: + if self._auth_thread is not None and self._auth_thread.isRunning(): + self._log.append( + 'Дождитесь завершения авторизации.' + ) + event.ignore() + return try: self._conn.line_received.disconnect(self._on_line_received) except (RuntimeError, TypeError): pass super().closeEvent(event) + + +class _TerminalHandshakeThread(QThread): + result_ready = Signal(object) + + def __init__(self, connection: ComConnection, parent: QWidget | None = None) -> None: + super().__init__(parent) + self._conn = connection + + def run(self) -> None: + try: + result = perform_handshake(self._conn, timeout=2.0) + except Exception as exc: + result = AuthResult(AuthStatus.FAILED, str(exc)) + self.result_ready.emit(result) diff --git a/app/widgets/dialogs/firmware_dialog.py b/app/widgets/dialogs/firmware_dialog.py index 2e63b96..a2bbfd1 100644 --- a/app/widgets/dialogs/firmware_dialog.py +++ b/app/widgets/dialogs/firmware_dialog.py @@ -36,6 +36,7 @@ from PySide6.QtWidgets import ( ) from ...services.com_service import ComConnection +from ...services.firmware_service import perform_authenticated_ota from ...ui.icons import get_icon_color, get_icon_muted, icon from .styled_dialog import StyledDialog @@ -243,61 +244,21 @@ class FirmwareDialog(StyledDialog): self._flash_btn.setEnabled(False) self._upload_btn.setEnabled(False) - self._log.append("Запуск OTA (04)…") + self._log.append("Запуск авторизованного OTA…") - # Очистить serial-буфер от остатков предыдущего опроса - try: - ser = self._conn.serial - if ser is not None and ser.is_open: - ser.reset_input_buffer() - except Exception: - pass + from PySide6.QtCore import QCoreApplication - # Приостанавливаем фоновый дренаж на время прошивки - acquired = self._conn._io_lock.acquire(timeout=0.5) - if not acquired: - self._log.append("✗ Ошибка: порт занят другой операцией") - self._flash_btn.setEnabled(True) - self._upload_btn.setEnabled(True) - return - - self._conn._cmd_owner = True - - try: - self._conn.send_command("04") - self._log.append("→ 04") - from PySide6.QtCore import QCoreApplication + def append_log(message: str) -> None: + self._log.append(message) QCoreApplication.processEvents() - # Контроллер шлёт прогресс 1-100, затем «Successful update» - # и через ~4 сек перезагружается с «READY!». - # Ждём до 60 сек (перепрошивка + перезагрузка). - deadline = time.time() + 60.0 - success = False - - while time.time() < deadline: - remaining = max(deadline - time.time(), 0.1) - resp = self._conn.read_response(timeout=min(remaining, 2.0)) - if resp is None: - QCoreApplication.processEvents() - continue - self._log.append(f"← {resp}") - QCoreApplication.processEvents() - if "ready" in resp.lower(): - success = True - break - - if success: - self._log.append("✓ Прошивка обновлена, плата перезагружена.") - else: - self._log.append("← таймаут (нет ответа)") - except Exception as exc: - self._log.append(f"Ошибка OTA: {exc}") - finally: - self._conn._cmd_owner = False - self._conn._io_lock.release() - self._flash_btn.setEnabled(True) - self._upload_btn.setEnabled(True) + result = perform_authenticated_ota(self._conn, log_message=append_log) + if result.success: + self._log.append(f"✓ {result.detail}.") + else: + self._log.append(f"✗ {result.detail}") + self._flash_btn.setEnabled(True) + self._upload_btn.setEnabled(True) def closeEvent(self, event) -> None: if self._upload_thread is not None and self._upload_thread.isRunning(): diff --git a/app/widgets/sections/firmware_section.py b/app/widgets/sections/firmware_section.py index c1b3f46..da2e052 100644 --- a/app/widgets/sections/firmware_section.py +++ b/app/widgets/sections/firmware_section.py @@ -23,6 +23,7 @@ from PySide6.QtWidgets import ( ) from ...services.com_service import ComConnection +from ...services.firmware_service import perform_authenticated_ota from ...ui.icons import get_icon_color, icon # Переиспользуем воркер из диалога @@ -195,36 +196,14 @@ class FirmwareSection(QWidget): self._pick_btn.setEnabled(True) self.upload_finished.emit() return - try: - self._conn.send_command("04") - self._log.append("→ 04") - - # Контроллер шлёт прогресс 1-100, затем «Successful update» - # и через ~4 сек перезагружается с «READY!». - # Ждём до 60 сек (перепрошивка + перезагрузка). - import time - deadline = time.time() + 60.0 - success = False - - while time.time() < deadline: - remaining = max(deadline - time.time(), 0.1) - resp = self._conn.read_response(timeout=min(remaining, 2.0)) - if resp is None: - continue - self._log.append(f"← {resp}") - if "ready" in resp.lower(): - success = True - break - - self._conn.release() - - if success: - self._log.append("✓ Прошивка обновлена, плата перезагружена.") - else: - self._log.append("← таймаут (нет ответа)") - except Exception as exc: - self._conn.release() - self._log.append(f"Ошибка OTA: {exc}") + result = perform_authenticated_ota( + self._conn, + log_message=self._log.append, + ) + if result.success: + self._log.append(f"✓ {result.detail}.") + else: + self._log.append(f"✗ {result.detail}") self._update_btn.setEnabled(True) self._pick_btn.setEnabled(True) self.upload_finished.emit() diff --git a/tests/test_firmware_service.py b/tests/test_firmware_service.py new file mode 100644 index 0000000..336b8d9 --- /dev/null +++ b/tests/test_firmware_service.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import unittest +from collections import deque +from unittest.mock import patch + +from app.services.auth_service import AuthStatus +from app.services.firmware_service import perform_authenticated_ota + + +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 + + 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 + + +class AuthenticatedOtaTests(unittest.TestCase): + _HOST_CHALLENGE = bytes.fromhex("00112233445566778899AABBCCDDEEFF") + _CHAL = "CHAL00112233445566778899AABBCCDDEEFF" + _RAES = "RAES40E9A18C22C2CFCD764666C90B358F9D" + + def _run(self, conn: FakeConnection): + with patch( + "app.services.auth_service.os.urandom", + return_value=self._HOST_CHALLENGE, + ): + return perform_authenticated_ota( + conn, + ota_timeout=0.1, + banner_timeout=0.1, + startup_delay=0.0, + handshake_timeout=0.1, + ready_timeout=0.1, + sleep_fn=lambda _seconds: None, + ) + + def test_authenticates_before_04_and_after_successful_update(self) -> None: + conn = FakeConnection([ + self._CHAL, + self._RAES, + "READY!", + "Writing 100%", + "Successful update", + "periphery controller ver_2.4.0 release", + self._CHAL, + self._RAES, + "READY!", + ]) + + result = self._run(conn) + + self.assertTrue(result.success) + self.assertEqual(AuthStatus.OK, result.pre_auth.status) + self.assertEqual(AuthStatus.OK, result.post_auth.status) + send_events = [event for event in conn.events if event.startswith("send:")] + self.assertEqual("send:04", send_events[1]) + self.assertTrue(send_events[0].startswith("send:07")) + self.assertTrue(send_events[2].startswith("send:07")) + + def test_does_not_send_04_when_pre_authentication_fails(self) -> None: + conn = FakeConnection([None]) + + result = self._run(conn) + + self.assertFalse(result.success) + self.assertEqual(AuthStatus.TIMEOUT, result.pre_auth.status) + self.assertFalse(any(event == "send:04" for event in conn.events)) + + def test_reports_post_authentication_failure_after_update(self) -> None: + conn = FakeConnection([ + self._CHAL, + self._RAES, + "READY!", + "Successful update", + None, + ]) + + result = self._run(conn) + + self.assertFalse(result.success) + self.assertEqual(AuthStatus.TIMEOUT, result.post_auth.status) + self.assertIn("после OTA", result.detail) + + +if __name__ == "__main__": + unittest.main()