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()