ayurishchevandClaude Opus 5.5
5f7bd6588a
Пункт 18 ревью 2026-09-28 21:32 (docs/changes/027): после отката ROS
плата остаётся на более новой прошивке, а приложение сравнивало версии
на неравенство и показывало откат как обновление.
- devices.fw_state: unknown/update/downgrade/current; has_fw_update и
фильтры не считают откат обновлением; метка «в ROS: X».
- upgrade_firmware не понижает прошивку; общая запись — _flash_firmware.
- Задача fw_downgrade: проверка → обязательный бэкап → запись →
перезагрузка; API POST /devices/{id}/firmware/downgrade и
/batch/fw_downgrade (target_version обязателен).
- UI: «Откатить прошивку…» (строка и группа), общее окно отката с kind.
Тесты: 39 из 39. Стенд: временное недоступное устройство — 422/202,
задачи failed на проверке, бэкапов нет. Реальная запись прошивки при
откате и ручная проверка UI не выполнялись.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
283 lines
17 KiB
Python
283 lines
17 KiB
Python
"""Операции над устройством через RouterOS REST. Не знают про БД и S3."""
|
|
import asyncio
|
|
import base64
|
|
import re
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from app.ros.client import RosClient, RosError
|
|
|
|
CHANNELS = ("stable", "long-term", "testing", "development")
|
|
|
|
|
|
_PRERELEASE_RE = re.compile(r"(alpha|beta|rc)", re.I)
|
|
|
|
|
|
def _ver_key(v: str | None) -> tuple:
|
|
head = re.match(r"\s*([\d.]+)", v or "") # числовая часть до суффикса: '7.25rc1' -> '7.25'
|
|
nums = [int(x) for x in re.findall(r"\d+", head.group(1) if head else "")][:4]
|
|
return tuple(nums + [0] * (4 - len(nums))) + (0 if _PRERELEASE_RE.search(v or "") else 1,)
|
|
|
|
|
|
def version_newer(latest: str | None, installed: str | None) -> bool:
|
|
"""True, если latest действительно новее installed. Строгое сравнение: на канале long-term «последняя»
|
|
версия может быть старше установленной — это не обновление. '7.25rc1' < '7.25'."""
|
|
return bool(latest and installed) and _ver_key(latest) > _ver_key(installed)
|
|
|
|
|
|
async def get_routerboard(c: RosClient) -> dict:
|
|
"""system/routerboard; на CHR и x86 раздела нет (HTTP 400 no such command) — это не сбой устройства."""
|
|
try:
|
|
return await c.get("system/routerboard") or {}
|
|
except RosError as e:
|
|
if e.status == 400 and "no such command" in str(e):
|
|
return {"routerboard": "false"}
|
|
raise
|
|
|
|
|
|
async def get_status(c: RosClient, check_updates: bool = True) -> dict:
|
|
"""Собирает статус: модель, канал, версии ROS/FW, uptime и доступные обновления.
|
|
|
|
latest-version в system/package/update появляется только после check-for-updates
|
|
(устройство обращается к серверам MikroTik). Сбой проверки не считается сбоем опроса.
|
|
"""
|
|
resource, board, update = await asyncio.gather(
|
|
c.get("system/resource"),
|
|
get_routerboard(c),
|
|
c.get("system/package/update"),
|
|
)
|
|
check_error = None
|
|
if check_updates:
|
|
try:
|
|
await c.post("system/package/update/check-for-updates", timeout=30)
|
|
update = await c.get("system/package/update")
|
|
except RosError as e:
|
|
check_error = str(e)
|
|
is_board = str(board.get("routerboard", "false")) == "true"
|
|
return {
|
|
"model": (board.get("model") if is_board else None) or resource.get("board-name"),
|
|
"channel": update.get("channel"),
|
|
"ros_version": resource.get("version"),
|
|
"ros_installed": update.get("installed-version"), # без суффикса «(stable)» — для сравнения
|
|
"ros_latest": update.get("latest-version"),
|
|
"ros_update_status": update.get("status"),
|
|
"ros_check_error": check_error,
|
|
"fw_current": board.get("current-firmware") if is_board else None,
|
|
"fw_upgrade": board.get("upgrade-firmware") if is_board else None,
|
|
"uptime": resource.get("uptime"),
|
|
}
|
|
|
|
|
|
async def wait_file(c: RosClient, name: str, attempts: int = 15) -> str:
|
|
"""Ждёт появления файла на устройстве и возвращает его .id."""
|
|
for _ in range(attempts):
|
|
files = await c.get("file", name=name)
|
|
if files:
|
|
return files[0][".id"]
|
|
await asyncio.sleep(1)
|
|
raise RosError(f"Файл {name} не появился на устройстве")
|
|
|
|
|
|
async def create_backup_files(c: RosClient, base: str) -> tuple[str, str]:
|
|
"""Создаёт бинарный бэкап (без шифрования) и .rsc-экспорт (с секретами). Возвращает имена файлов."""
|
|
binary, rsc = f"{base}.backup", f"{base}.rsc"
|
|
await c.post("system/backup/save", {"name": base, "dont-encrypt": "yes"})
|
|
# show-sensitive: в .rsc попадают пароли, ключи и секреты — из файла можно восстановить всё, что их использует
|
|
await c.post("export", {"file": base, "show-sensitive": ""})
|
|
await wait_file(c, binary)
|
|
await wait_file(c, rsc)
|
|
return binary, rsc
|
|
|
|
|
|
async def download_file(c: RosClient, name: str, dest: Path, chunk_size: int = 32768) -> int:
|
|
"""Скачивает файл блоками через REST (/file/read). REST не отдаёт содержимое больших файлов
|
|
через GET /file, а бинарные данные в JSON ненадёжны, поэтому блок читается скриптом и
|
|
передаётся в base64. Возвращает размер файла; сверяет его с размером на устройстве."""
|
|
files = await c.get("file", name=name)
|
|
if not files:
|
|
raise RosError(f"Файл {name} не найден на устройстве")
|
|
size = int(files[0]["size"])
|
|
received = 0
|
|
with dest.open("wb") as f:
|
|
while received < size:
|
|
script = (f':put [:convert to=base64 ([/file/read file="{name}" offset={received} '
|
|
f'chunk-size={chunk_size} as-value]->"data")]')
|
|
res = await c.post("execute", {"script": script, "as-string": ""})
|
|
chunk = base64.b64decode((res or {}).get("ret", ""))
|
|
if not chunk:
|
|
raise RosError(f"Пустой блок при чтении {name} со смещения {received}")
|
|
f.write(chunk)
|
|
received += len(chunk)
|
|
if received != size:
|
|
raise RosError(f"Размер {name}: получено {received} из {size} байт")
|
|
return size
|
|
|
|
|
|
async def remove_file(c: RosClient, filename: str) -> None:
|
|
files = await c.get("file", name=filename)
|
|
for f in files or []:
|
|
await c.delete(f"file/{f['.id']}")
|
|
|
|
|
|
async def set_channel(c: RosClient, channel: str) -> None:
|
|
if channel not in CHANNELS:
|
|
raise ValueError(f"Неизвестный канал: {channel}")
|
|
await c.post("system/package/update/set", {"channel": channel})
|
|
|
|
|
|
async def _check_update(c: RosClient) -> tuple[str | None, str | None]:
|
|
"""check-for-updates + чтение system/package/update. Возвращает (installed, latest)."""
|
|
await c.post("system/package/update/check-for-updates", timeout=120)
|
|
u = await c.get("system/package/update")
|
|
return u.get("installed-version"), u.get("latest-version")
|
|
|
|
|
|
async def _install(c: RosClient) -> None:
|
|
"""POST system/package/update/install: после скачивания устройство уходит в перезагрузку
|
|
и рвёт соединение — это штатно (RosError со status=None)."""
|
|
try:
|
|
await c.post("system/package/update/install", timeout=15)
|
|
except RosError as e:
|
|
if e.status is not None:
|
|
raise
|
|
|
|
|
|
async def install_ros_update(c: RosClient) -> str:
|
|
"""Проверяет обновления и запускает установку. Устройство само перезагрузится после скачивания."""
|
|
installed, latest = await _check_update(c)
|
|
if not version_newer(latest, installed):
|
|
return f"Обновление не требуется (установлена {installed}, на канале {latest})"
|
|
await _install(c)
|
|
return f"Обновление {installed} → {latest} запущено, устройство перезагрузится"
|
|
|
|
|
|
async def check_downgrade(c: RosClient, target_version: str) -> tuple[str | None, str | None]:
|
|
"""Проверяет, что версия канала совпадает с подтверждённой пользователем и действительно старше
|
|
установленной (иначе это не откат). Install не выполняет — чтобы не делать бэкап устройства,
|
|
которое откатывать нельзя. Возвращает (installed, latest)."""
|
|
installed, latest = await _check_update(c)
|
|
if latest != target_version:
|
|
raise RosError(f"Версия канала {latest} не совпадает с подтверждённой {target_version} — откат отменён")
|
|
if not version_newer(installed, latest):
|
|
raise RosError(f"Версия канала {latest} не старше установленной {installed} — это не откат")
|
|
return installed, latest
|
|
|
|
|
|
async def downgrade_ros(c: RosClient, target_version: str) -> str:
|
|
"""Осознанный откат ROS до версии канала (long-term/testing старше установленной): RouterOS 7 при
|
|
/system/package/update/install скачивает версию канала и выполняет откат, если она старше установленной."""
|
|
installed, latest = await check_downgrade(c, target_version)
|
|
await _install(c)
|
|
return f"Откат {installed} → {latest} запущен, устройство перезагрузится"
|
|
|
|
|
|
FW_DONE_MSG = "Firmware upgraded successfully, please reboot for changes to take effect!"
|
|
_UPTIME_RE = re.compile(r"^(?:(\d+)w)?(?:(\d+)d)?(?:(\d+)h)?(?:(\d+)m)?(?:(\d+)s)?$")
|
|
|
|
|
|
def parse_uptime(value: str | None) -> int | None:
|
|
"""'1w2d3h4m5s' -> секунды; None, если формат не распознан."""
|
|
m = _UPTIME_RE.match(value or "")
|
|
if not m or not any(m.groups()):
|
|
return None
|
|
w, d, h, mi, sec = (int(x or 0) for x in m.groups())
|
|
return (((w * 7 + d) * 24 + h) * 60 + mi) * 60 + sec
|
|
|
|
|
|
def _ts(value: str | None) -> datetime | None:
|
|
try:
|
|
return datetime.strptime(value or "", "%Y-%m-%d %H:%M:%S")
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
async def _fw_done_entries(c: RosClient) -> list[dict]:
|
|
# фильтр REST ?message=… по точному тексту ничего не возвращает — фильтруем сами
|
|
log = await c.get("log", **{".proplist": ".id,time,message"})
|
|
return [e for e in log or [] if e.get("message") == FW_DONE_MSG]
|
|
|
|
|
|
async def _booted_at(c: RosClient) -> datetime | None:
|
|
"""Время последней загрузки устройства по его часам (в той же зоне, что и время журнала)."""
|
|
clock, res = await asyncio.gather(c.get("system/clock"), c.get("system/resource"))
|
|
now = _ts(f"{clock.get('date')} {clock.get('time')}")
|
|
up = parse_uptime(res.get("uptime"))
|
|
return now - timedelta(seconds=up) if now and up is not None else None
|
|
|
|
|
|
async def _flash_firmware(c: RosClient, wait_seconds: int, poll: float) -> str | None:
|
|
"""Записывает встроенную в ROS прошивку (POST system/routerboard/upgrade) и ждёт появления FW_DONE_MSG
|
|
в журнале, либо обнаруживает, что запись уже была сделана после последней загрузки (изменение 027: общая
|
|
часть upgrade_firmware и downgrade_firmware — команда и признак завершения одни и те же, отличается только
|
|
итоговое сообщение). Возвращает готовое сообщение для уже записанной прошивки, иначе None — сообщение о
|
|
свежей записи формирует вызывающий код (текст различается для обновления и отката)."""
|
|
before = await _fw_done_entries(c)
|
|
booted = await _booted_at(c)
|
|
if booted and any((t := _ts(e.get("time"))) and t >= booted for e in before):
|
|
# прошивка уже записана после последней загрузки, ждём только перезагрузки
|
|
await _reboot(c)
|
|
return "Прошивка уже записана (запись в журнале), перезагрузка отправлена"
|
|
|
|
known = {e[".id"] for e in before}
|
|
await c.post("system/routerboard/upgrade")
|
|
for _ in range(int(wait_seconds / poll) if poll else 1000):
|
|
if any(e[".id"] not in known for e in await _fw_done_entries(c)):
|
|
await _reboot(c)
|
|
return None
|
|
await asyncio.sleep(poll)
|
|
raise RosError("В журнале не появилась запись об обновлении прошивки за отведённое время")
|
|
|
|
|
|
async def upgrade_firmware(c: RosClient, wait_seconds: int = 180, poll: float = 2.0) -> str:
|
|
"""Обновляет FW и перезагружает устройство, как только в журнале появилась запись
|
|
«Firmware upgraded successfully…». current-firmware до перезагрузки не меняется,
|
|
поэтому по нему завершение определить нельзя."""
|
|
board = await get_routerboard(c)
|
|
if str(board.get("routerboard", "false")) != "true":
|
|
return "Устройство без RouterBOARD firmware (например, CHR) — пропущено"
|
|
current, upgrade = board.get("current-firmware"), board.get("upgrade-firmware")
|
|
if current == upgrade:
|
|
return f"Прошивка актуальна ({current})"
|
|
if version_newer(current, upgrade):
|
|
# изменение 027: плата записана прошивкой новее встроенной в установленную ROS (например, после отката
|
|
# ROS) — это не обновление, а откат; штатное обновление его не выполняет
|
|
return (f"Прошивка платы {current} новее встроенной в ROS {upgrade} — обновление не требуется; "
|
|
f"откат — «Откатить прошивку…»")
|
|
|
|
already = await _flash_firmware(c, wait_seconds, poll)
|
|
return already or f"Прошивка обновлена до {upgrade}, перезагрузка отправлена"
|
|
|
|
|
|
async def check_fw_downgrade(c: RosClient, target_version: str) -> tuple[str | None, str | None]:
|
|
"""Изменение 027: проверяет осознанный откат прошивки RouterBOARD до версии, встроенной в установленную ROS —
|
|
по образцу check_downgrade. Не RouterBOARD — откат недоступен; встроенная версия должна совпадать с
|
|
подтверждённой пользователем и быть действительно старше записанной на плате (иначе это не откат). Запись не
|
|
выполняет — чтобы не делать бэкап устройства, которое откатывать нельзя. Возвращает (current, upgrade)."""
|
|
board = await get_routerboard(c)
|
|
if str(board.get("routerboard", "false")) != "true":
|
|
raise RosError("Устройство без RouterBOARD firmware (например, CHR) — откат прошивки недоступен")
|
|
current, upgrade = board.get("current-firmware"), board.get("upgrade-firmware")
|
|
if upgrade != target_version:
|
|
raise RosError(f"Прошивка в ROS {upgrade} не совпадает с подтверждённой {target_version} — откат отменён")
|
|
if not version_newer(current, upgrade):
|
|
raise RosError(f"Прошивка в ROS {upgrade} не старше записанной {current} — это не откат")
|
|
return current, upgrade
|
|
|
|
|
|
async def downgrade_firmware(c: RosClient, target_version: str, wait_seconds: int = 180, poll: float = 2.0) -> str:
|
|
"""Изменение 027: осознанный откат прошивки RouterBOARD до версии, встроенной в установленную ROS (после
|
|
отката ROS, например изменение 023, плата остаётся на прежней прошивке). Технически — та же команда
|
|
/system/routerboard/upgrade, что и обычное обновление: она записывает версию, встроенную в установленную
|
|
ROS, в том числе более старую."""
|
|
current, upgrade = await check_fw_downgrade(c, target_version)
|
|
already = await _flash_firmware(c, wait_seconds, poll)
|
|
return already or f"Откат прошивки {current} → {upgrade} записан, перезагрузка отправлена"
|
|
|
|
|
|
async def _reboot(c: RosClient) -> None:
|
|
try:
|
|
await c.post("system/reboot", timeout=10)
|
|
except RosError as e:
|
|
if e.status is not None: # обрыв соединения при перезагрузке — штатно
|
|
raise
|