feat: add authenticated firmware update flow
- добавлена полная повторная авторизация перед командой 04 - добавлена авторизация после успешного обновления и перезагрузки платы - общий OTA-сценарий подключён к базовому и PRO-интерфейсам - в терминал команд добавлена кнопка ручной авторизации - состояние авторизации синхронизировано с основной вкладкой - добавлены тесты порядка авторизации и обработки ошибок OTA - версия приложения повышена до 1.5.0 - обновлена документация
This commit is contained in:
parent
5513eb8a56
commit
c34cb0cd7e
@ -118,6 +118,9 @@
|
||||
- Зеленый — подключено, плата готова к работе.
|
||||
|
||||
Кнопка переподключения (↻) позволяет быстро восстановить связь при обрыве.
|
||||
Перед запуском OTA и после успешного обновления приложение автоматически
|
||||
выполняет полную AES-256 авторизацию платы. В терминале команд доступна
|
||||
отдельная кнопка ручной авторизации.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -1,3 +1,3 @@
|
||||
"""Plate Config App package."""
|
||||
|
||||
__version__ = "1.4.2"
|
||||
__version__ = "1.5.0"
|
||||
|
||||
119
app/services/firmware_service.py
Normal file
119
app/services/firmware_service.py
Normal file
@ -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()
|
||||
@ -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, "В разработке", "Эта функция пока не реализована.")
|
||||
|
||||
|
||||
@ -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('<span style="color:#f87171;">✗ Порт не подключён</span>')
|
||||
return
|
||||
if self._auth_thread is not None:
|
||||
return
|
||||
|
||||
self._set_command_controls_enabled(False)
|
||||
self._auth_btn.setText("Авторизация…")
|
||||
self._log.append('<span style="color:#facc15;">Запуск полной авторизации…</span>')
|
||||
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(
|
||||
'<span style="color:#4ade80;">✓ Авторизация успешно завершена</span>'
|
||||
)
|
||||
else:
|
||||
self._log.append(
|
||||
f'<span style="color:#f87171;">✗ Авторизация: {result.detail}</span>'
|
||||
)
|
||||
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'<span style="color:#93c5fd;">← {stripped}</span>')
|
||||
|
||||
def closeEvent(self, event) -> None:
|
||||
if self._auth_thread is not None and self._auth_thread.isRunning():
|
||||
self._log.append(
|
||||
'<span style="color:#facc15;">Дождитесь завершения авторизации.</span>'
|
||||
)
|
||||
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)
|
||||
|
||||
@ -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():
|
||||
|
||||
@ -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()
|
||||
|
||||
102
tests/test_firmware_service.py
Normal file
102
tests/test_firmware_service.py
Normal file
@ -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()
|
||||
Loading…
x
Reference in New Issue
Block a user