Danil Iarosh
add8aff0af
- Добавлена обратная совместимость с прошивками, содержащими маркер no_auth. - Повторная авторизация после команды 04 и успешного обновления прошивки приведена к требуемому порядку. - В терминал команд добавлена ручная кнопка авторизации. - Реализовано OTA-обновление приложения с загрузкой платформенного артефакта, безопасной заменой и перезапуском. - Проверка обновлений учитывает только более новые версии по схеме major, minor и patch. - Обновлены имена Windows и Linux сборок, документация и тестовое покрытие.
307 lines
12 KiB
Python
307 lines
12 KiB
Python
"""Запуск и AES-256 авторизация платы периферийного контроллера.
|
||
|
||
Стартовая последовательность:
|
||
1. Плата → компьютер: periphery controller ver_<версия и комментарий>
|
||
2. Компьютер ждёт 2000 мс.
|
||
3. Компьютер → плата: 07<32 hex> (16 случайных байт)
|
||
4. Плата → компьютер: CHAL<32 hex> (встречный запрос, 16 байт)
|
||
5. Компьютер → плата: 08<32 hex> (зашифрованный ответ на CHAL)
|
||
6. Плата → компьютер: RAES<32 hex> (ответ на запрос из шага 3)
|
||
7. Плата → компьютер: READY!
|
||
|
||
Авторизация считается завершённой только после точной строки ``READY!``.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hmac
|
||
import os
|
||
import time
|
||
from collections.abc import Callable
|
||
from dataclasses import dataclass
|
||
from enum import Enum, auto
|
||
|
||
from Crypto.Cipher import AES
|
||
|
||
from .com_service import ComConnection
|
||
|
||
STARTUP_BANNER_PREFIX = "periphery controller ver_"
|
||
NO_AUTH_MARKER = "no_auth"
|
||
STARTUP_DELAY_SECONDS = 2.0
|
||
STARTUP_BANNER_TIMEOUT_SECONDS = 20.0
|
||
READY_TIMEOUT_SECONDS = 5.0
|
||
|
||
# Ключ хранится в том же обфусцированном виде, что и в рабочем клиенте.
|
||
_AES_KEY_MASK = 0x37
|
||
_AES_OBFUSCATED_KEY = bytes((
|
||
0x69, 0x40, 0xC7, 0x0E, 0xF2, 0xC7, 0x45, 0x70,
|
||
0x72, 0xCD, 0x28, 0x58, 0x99, 0x8F, 0x73, 0xA8,
|
||
0x5F, 0x5F, 0xED, 0xB4, 0xC1, 0x62, 0xB4, 0x24,
|
||
0xAC, 0xD7, 0xA3, 0xD6, 0xD8, 0x0B, 0x50, 0x37,
|
||
))
|
||
_AES_KEY = bytes(value ^ _AES_KEY_MASK for value in _AES_OBFUSCATED_KEY)
|
||
|
||
|
||
class AuthStatus(Enum):
|
||
OK = auto()
|
||
NEED_AUTH = auto()
|
||
FAILED = auto()
|
||
TIMEOUT = auto()
|
||
DISCONNECTED = auto()
|
||
|
||
|
||
@dataclass
|
||
class AuthResult:
|
||
status: AuthStatus
|
||
detail: str = ""
|
||
startup_ready: bool = False
|
||
firmware_version: str | None = None
|
||
auth_skipped: bool = False
|
||
|
||
|
||
def firmware_requires_auth(version: str) -> bool:
|
||
"""Определить режим авторизации по строке версии прошивки."""
|
||
return NO_AUTH_MARKER not in version.casefold()
|
||
|
||
|
||
def _remember_firmware(conn: ComConnection, banner: str) -> str:
|
||
version = banner[len(STARTUP_BANNER_PREFIX):].strip()
|
||
conn.set_firmware_version(
|
||
version,
|
||
auth_required=firmware_requires_auth(version),
|
||
)
|
||
return version
|
||
|
||
|
||
def _aes_encrypt(plaintext: bytes, key: bytes = _AES_KEY) -> bytes:
|
||
"""ECB-шифрование одного 16-байтового блока."""
|
||
cipher = AES.new(key, AES.MODE_ECB)
|
||
return cipher.encrypt(plaintext)
|
||
|
||
|
||
def _aes_decrypt(ciphertext: bytes, key: bytes = _AES_KEY) -> bytes:
|
||
cipher = AES.new(key, AES.MODE_ECB)
|
||
return cipher.decrypt(ciphertext)
|
||
|
||
|
||
def _read_until(
|
||
conn: ComConnection,
|
||
predicate: Callable[[str], bool],
|
||
timeout: float,
|
||
) -> str | None:
|
||
"""Читать полные строки до совпадения или истечения общего таймаута."""
|
||
deadline = time.monotonic() + timeout
|
||
while conn.is_connected:
|
||
remaining = deadline - time.monotonic()
|
||
if remaining <= 0:
|
||
return None
|
||
line = conn.read_response(timeout=remaining)
|
||
if line is None:
|
||
return None
|
||
if predicate(line.strip()):
|
||
return line.strip()
|
||
return None
|
||
|
||
|
||
def wait_for_startup_banner(
|
||
conn: ComConnection,
|
||
timeout: float = STARTUP_BANNER_TIMEOUT_SECONDS,
|
||
) -> str | None:
|
||
"""Дождаться ``periphery controller ver_<...>`` с непустым суффиксом."""
|
||
def is_banner(line: str) -> bool:
|
||
if not line.startswith(STARTUP_BANNER_PREFIX):
|
||
return False
|
||
return bool(line[len(STARTUP_BANNER_PREFIX):].strip())
|
||
|
||
return _read_until(conn, is_banner, timeout)
|
||
|
||
|
||
def check_status(conn: ComConnection, timeout: float = 3.0) -> AuthStatus:
|
||
"""Отправить команду 99 и проанализировать ответ.
|
||
|
||
Для прошивки с авторизацией только точная строка ``READY!`` означает
|
||
готовность. Для обнаруженной прошивки ``no_auth`` достаточно любого
|
||
непустого ответа: отсутствие ответа считается потерей связи, а не
|
||
запросом авторизации.
|
||
"""
|
||
if not conn.is_connected:
|
||
return AuthStatus.DISCONNECTED
|
||
try:
|
||
conn.send_command("99")
|
||
except Exception:
|
||
return AuthStatus.DISCONNECTED
|
||
|
||
try:
|
||
if conn.auth_required is False:
|
||
response = _read_until(conn, bool, timeout)
|
||
return AuthStatus.OK if response is not None else AuthStatus.DISCONNECTED
|
||
|
||
ready = _read_until(conn, lambda line: line == "READY!", timeout)
|
||
return AuthStatus.OK if ready is not None else AuthStatus.NEED_AUTH
|
||
finally:
|
||
conn.release()
|
||
|
||
|
||
def perform_required_handshake(
|
||
conn: ComConnection,
|
||
timeout: float = 2.0,
|
||
ready_timeout: float = READY_TIMEOUT_SECONDS,
|
||
) -> AuthResult:
|
||
"""Выполнить handshake только если он требуется текущей прошивке."""
|
||
if not conn.is_connected:
|
||
return AuthResult(AuthStatus.DISCONNECTED, "Порт не открыт")
|
||
if conn.auth_required is False:
|
||
version = conn.firmware_version
|
||
detail = "Авторизация не требуется: обнаружена прошивка no_auth"
|
||
if version:
|
||
detail = f"Прошивка {version}: авторизация не требуется"
|
||
return AuthResult(
|
||
AuthStatus.OK,
|
||
detail,
|
||
firmware_version=version,
|
||
auth_skipped=True,
|
||
)
|
||
return perform_handshake(
|
||
conn,
|
||
timeout=timeout,
|
||
ready_timeout=ready_timeout,
|
||
)
|
||
|
||
|
||
def perform_handshake(
|
||
conn: ComConnection,
|
||
timeout: float = 2.0,
|
||
ready_timeout: float = READY_TIMEOUT_SECONDS,
|
||
) -> AuthResult:
|
||
"""Выполнить полный цикл двойного рукопожатия.
|
||
|
||
Returns:
|
||
AuthResult с итоговым статусом.
|
||
"""
|
||
if not conn.is_connected:
|
||
return AuthResult(AuthStatus.DISCONNECTED, "Порт не открыт")
|
||
|
||
# ── Шаг 1: генерируем 16 случайных байт и отправляем 07<hex> ──
|
||
our_challenge = os.urandom(16)
|
||
our_challenge_hex = our_challenge.hex().upper()
|
||
cmd_07 = "07" + our_challenge_hex
|
||
try:
|
||
conn.send_command(cmd_07)
|
||
except Exception as exc:
|
||
conn.release()
|
||
return AuthResult(AuthStatus.FAILED, f"Ошибка отправки 07: {exc}")
|
||
|
||
try:
|
||
# ── Шаг 2: ждём CHAL<32 hex> от контроллера ──
|
||
resp = _read_until(conn, lambda line: line.upper().startswith("CHAL"), timeout)
|
||
if resp is None:
|
||
return AuthResult(AuthStatus.TIMEOUT, "Таймаут ожидания CHAL от контроллера")
|
||
|
||
# Парсим CHAL — допускаем разделители (пробел, двоеточие)
|
||
board_challenge_hex = resp[4:].strip().lstrip(":").strip()
|
||
if len(board_challenge_hex) != 32:
|
||
return AuthResult(
|
||
AuthStatus.FAILED,
|
||
f"Некорректная длина CHAL: {board_challenge_hex}",
|
||
)
|
||
|
||
try:
|
||
board_challenge = bytes.fromhex(board_challenge_hex)
|
||
except ValueError:
|
||
return AuthResult(
|
||
AuthStatus.FAILED,
|
||
f"CHAL содержит не-hex символы: {board_challenge_hex}",
|
||
)
|
||
|
||
# ── Шаг 3: шифруем board_challenge AES-256 и отправляем 08<hex> ──
|
||
encrypted_response = _aes_encrypt(board_challenge)
|
||
cmd_08 = "08" + encrypted_response.hex().upper()
|
||
try:
|
||
conn.write_raw(cmd_08)
|
||
except Exception as exc:
|
||
return AuthResult(AuthStatus.FAILED, f"Ошибка отправки 08: {exc}")
|
||
|
||
# ── Шаг 4: проверяем доказательство контроллера RAES ──
|
||
raes = _read_until(conn, lambda line: line.upper().startswith("RAES"), timeout)
|
||
if raes is None:
|
||
return AuthResult(AuthStatus.TIMEOUT, "Таймаут ожидания RAES от контроллера")
|
||
|
||
controller_proof_hex = raes[4:].strip().lstrip(":").strip()
|
||
try:
|
||
controller_proof = bytes.fromhex(controller_proof_hex)
|
||
except ValueError:
|
||
return AuthResult(AuthStatus.FAILED, "RAES содержит некорректные hex-данные")
|
||
|
||
expected_proof = _aes_encrypt(our_challenge)
|
||
if len(controller_proof) != 16 or not hmac.compare_digest(
|
||
controller_proof,
|
||
expected_proof,
|
||
):
|
||
return AuthResult(AuthStatus.FAILED, "Проверка RAES контроллера не пройдена")
|
||
|
||
# AUTH_PC_OK и другие промежуточные строки игнорируются. Успех —
|
||
# только точная отдельная строка READY! после RAES.
|
||
ready = _read_until(conn, lambda line: line == "READY!", ready_timeout)
|
||
if ready is None:
|
||
return AuthResult(AuthStatus.TIMEOUT, "Таймаут ожидания READY! от контроллера")
|
||
|
||
return AuthResult(AuthStatus.OK, "Авторизация подтверждена строкой READY!")
|
||
finally:
|
||
conn.release()
|
||
|
||
|
||
def perform_startup_handshake(
|
||
conn: ComConnection,
|
||
*,
|
||
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,
|
||
) -> AuthResult:
|
||
"""Дождаться загрузки и выполнить handshake, если он требуется."""
|
||
if not conn.is_connected:
|
||
return AuthResult(AuthStatus.DISCONNECTED, "Порт не открыт")
|
||
|
||
try:
|
||
banner = wait_for_startup_banner(conn, timeout=banner_timeout)
|
||
except Exception as exc:
|
||
return AuthResult(AuthStatus.FAILED, f"Ошибка ожидания строки запуска: {exc}")
|
||
|
||
if banner is None:
|
||
return AuthResult(
|
||
AuthStatus.TIMEOUT,
|
||
f"Таймаут ожидания {STARTUP_BANNER_PREFIX}<версия>",
|
||
)
|
||
|
||
version = _remember_firmware(conn, banner)
|
||
sleep_fn(startup_delay)
|
||
if not conn.is_connected:
|
||
return AuthResult(
|
||
AuthStatus.DISCONNECTED,
|
||
"Порт закрыт до проверки режима авторизации",
|
||
startup_ready=True,
|
||
firmware_version=version,
|
||
)
|
||
|
||
if not firmware_requires_auth(version):
|
||
conn.release()
|
||
return AuthResult(
|
||
AuthStatus.OK,
|
||
f"{banner}\nАвторизация не требуется для прошивки no_auth",
|
||
startup_ready=True,
|
||
firmware_version=version,
|
||
auth_skipped=True,
|
||
)
|
||
|
||
result = perform_handshake(
|
||
conn,
|
||
timeout=handshake_timeout,
|
||
ready_timeout=ready_timeout,
|
||
)
|
||
result.startup_ready = True
|
||
result.firmware_version = version
|
||
if result.status == AuthStatus.OK:
|
||
result.detail = f"{banner}\n{result.detail}"
|
||
return result
|