Skip to content
BranchesTags

Repository files navigation

MikroTik Backup

Скрипт для автоматизированного сбора конфигураций с устройств MikroTik RouterOS через SSH и сохранения их в Git репозиторий.

Особенности

  • Экспортируемый конфиг не сохраняется на самом роутере
  • Параллельное подключение к нескольким устройствам
  • Аутентификация по SSH-ключам с проверкой прав доступа
  • Поддержка экспорта чувствительных данных (пароли, ключи)
  • Потокобезопасное логирование
  • Автоматический commit и push в Git
  • Определение изменений в конфигурациях через diff
  • Автоматическая очистка временных файлов

Требования

  • Python 3.12 или выше
  • SSH доступ к устройствам MikroTik
  • Git репозиторий для хранения бэкапов

Подготовка mikrotik

  1. Создайте группу с права для получения резервных копий, например backup:
/user group add name=backup policy="ssh,read,sensitive,!local,!telnet,!ftp,\
    !reboot,!write,!policy,!test,!winbox,!password,!web,!sniff,!api,!romon,!rest-api"
  1. Добавьте пользователя backup, назначте его в группу backup:
 /user add name=backup group=backup comment=backup
 /user expire-password backup
  1. Создайте или используйте существующий, нужно добавить открытую часть rsa ключа пользователю backup:
/user ssh-keys add user=backup key="ssh-rsa AAAA..."

Установка

  1. Клонирование репозитория:
git clone https://github.com/naem53/mikrotik-backup.git
cd mikrotik-backup
  1. Установка глобальных зависимостей:
pip install --upgrade poetry poetry-dotenv-plugin
  1. Установка зависимостей проекта:
poetry install

Запуск без использования docker

Запуск:

poetry run backup

Запуск через docker

Измените интервал/переодичность запуска скрипта cron в docker/crontabs/root.

В docker/ размещены Dockerfile и Dockerfile.ca:

  • Dockerfile - интрукция сборки без собственных сертификатов
  • Dockerfile.ca - инструкция сборки с self-managed сертификатом, но нужно сертификат или цепочку сертификатов положить в корень проекта для добавления его в image.

Перед запуском измените в docker-compose.yaml Dockerfile на нужный Вам.

Запуск:

  1. Установите docker.

  2. Запускаем через docker compose:

docker compose up -d --build
  1. Смотрим логи:
docker compose logs -f

Конфигурация

Создайте файл config.yaml, для примера используйте файл config.example.yaml:

git:
  url: https://github.com/user/backups.git
  branch: main
  username: your-github-username
  password: your-github-token
  path: /tmp/mikrotik_backups
  commit_message: "Update backups for hosts:"

parallel:
  max_workers: 10
  timeout_per_host: 60

telegram:
  enabled: true
  bot_token: "1234567890:ABCdefGHIJKLMNOPQRSTUVWXYZ"  # Токен вашего бота от @BotFather
  chat_id: "-34534534534"      # ID чата или пользователя для отправки уведомлений
  notify_on_success: true      # Уведомлять всегда (даже если нет ошибок и изменений)
  notify_on_changes: true      # Уведомлять об изменениях
  notify_on_failure: true      # Уведомлять об ошибках
  parse_mode: "html"           # "html" или "markdown"
  # additional_chat_ids:         # Опционально: дополнительные получатели
  #   - "567563345"
  #   - "555555555"

email:
  enabled: true
  smtp_host: smtp.mail.com
  smtp_port: 465
  username: "mail"
  password: "password"
  from_email: "backup@mail.com"
  to_emails:
    - "admin@example.com"
    - "it@example.com"
  use_ssl: true
  use_tls: false
  notify_on_success: true
  notify_on_changes: true
  notify_on_failure: true

defaults:
  username: admin
  key_file: /home/user/.ssh/id_rsa
  port: 22
  sensitive: false
  timeout: 30

hosts:
  - hostname: 192.168.1.1
  - hostname: 192.168.1.2
    username: backup_user
    port: 2222
    sensitive: true

Пример вывода:

INFO: Начинаем загрузку конфигурации из yaml файла: config.yaml
INFO: Конфигурация успешно загружена.
INFO: Клонирование репозитория в /tmp/mikrotik_backups
INFO: Репозиторий успешно клонирован
INFO: Подключение к 192.168.1.1:22 под пользователем: admin
INFO: Успешно подключен к 192.168.1.1
INFO: Успешно получена конфигурация (15234 байт)
INFO: Хост 192.168.1.1 успешно обработан
INFO: Обработка завершена. Успешно: 2, Ошибок: 0
INFO: Коммитим изменения
INFO: Изменения успешно запушены
INFO: Директория удалена

Описание классов, методов и функций скрипта

Модуль конфигураций.

EmailConfig Objects

@dataclass(slots=True)
class EmailConfig()

Конфигурация Email уведомлений.

TelegramConfig Objects

@dataclass
class TelegramConfig()

Конфигурация Telegram уведомлений.

Attributes:

  • enabled - Включены ли уведомления
  • bot_token - Токен Telegram бота
  • chat_id - ID чата для отправки
  • chat_name - Имя чата (загружается автоматически)
  • notify_on_success - Уведомлять об успешном выполнении
  • notify_on_failure - Уведомлять об ошибках
  • notify_on_changes - Уведомлять об изменениях
  • parse_mode - Формат сообщения ("html" или "markdown")
  • additional_chat_ids - Дополнительные ID чатов

GetBackupConfig Objects

@dataclass
class GetBackupConfig()

Конфигурация подключения для получения резервной копии через get_backup().

Attributes:

  • hostname - Адрес хоста (IP или доменное имя) для подключения.
  • username - Имя пользователя для аутентификации на удалённом хосте.
  • key_file - Путь к файлу приватного SSH-ключа.
  • port - Порт для SSH-подключения (по умолчанию 22).
  • sensitive - Флаг, указывающий, содержит ли параметр чувствительные данные в экспортируемом конфиге. По умолчанию False.
  • timeout - Таймаут подключения в секундах (по умолчанию 30).

.

EmailNotifier Objects

class EmailNotifier()

.

__init__

def __init__(config: EmailConfig) -> None

.

send_message

def send_message(subject: str, message: str) -> bool

Отправка email сообщения.

build_notification_message

def build_notification_message(results: dict,
                               changed_hosts: list[str]) -> tuple[str, str]

Формирует тему и текст уведомления.

should_notify

def should_notify(results: dict, changed_hosts: list[str]) -> bool

Проверяет нужно ли отправлять уведомление.

send_complete_notification

def send_complete_notification(results: dict,
                               changed_hosts: list[str]) -> bool

Отправка полного уведомления.

Модуль потокобезопасного логирования.

Предоставляет централизованную настройку системы логирования и вспомогательную функцию safe_log для безопасной записи логов из многопоточной среды.

Основные возможности:

  • настройка глобального форматирования логов;
  • поддержка уровней INFO, WARNING и ERROR;
  • потокобезопасная запись сообщений через threading.Lock;
  • подавление избыточных логов от paramiko.transport;
  • единый интерфейс логирования для всего приложения.

Модуль используется для:

  • логирования операций резервного копирования;
  • записи ошибок сетевых подключений;
  • отображения статуса Git-операций;
  • логирования Telegram-уведомлений;
  • диагностики выполнения приложения.

Functions: safe_log: Потокобезопасная запись сообщений в лог.

Globals: logger: Экземпляр стандартного Python logger для текущего модуля.

log_lock: Блокировка для синхронизации записи логов между потоками.

Notes:

  • Используется стандартный модуль logging.
  • Формат логов включает дату, время, уровень и сообщение.
  • Логи paramiko.transport понижены до уровня WARNING для уменьшения количества служебных сообщений.
  • Все операции логирования выполняются синхронно.

Example:

safe_log("info", "Приложение успешно запущено") safe_log("warning", "Повторная попытка подключения") safe_log("error", "Ошибка выполнения операции")

safe_log

def safe_log(level: Literal["info", "error", "warning"], message: str) -> None

Потокобезопасное логирование сообщений с указанным уровнем.

Функция использует блокировку (lock) для предотвращения состояний гонки при одновременной записи в лог из нескольких потоков.

Arguments:

  • level - Уровень логирования. Допустимые значения:
    • "info" - информационное сообщение
    • "error" - сообщение об ошибке
    • "warning" - предупреждение
  • message - Текст сообщения для записи в лог.

Returns:

  • None - Функция ничего не возвращает.

Raises:

  • ValueError - Если передан неподдерживаемый уровень логирования.

Example:

safe_log("info", "Сервер успешно запущен") safe_log("error", "Не удалось подключиться к базе данных")

Модуль для управления Git-репозиториями.

Содержит реализацию класса GitManager, предназначенного для автоматизации работы с Git через библиотеку GitPython.

Модуль предоставляет функциональность для:

  • клонирования удалённых репозиториев;
  • создания и переключения веток;
  • настройки upstream-веток;
  • добавления файлов в индекс Git;
  • создания коммитов;
  • выполнения push и pull операций;
  • определения изменённых конфигурационных файлов;
  • получения статуса репозитория;
  • очистки локальной копии репозитория.

Поддерживаются:

  • HTTPS-репозитории;
  • аутентификация по логину и паролю/token;
  • работа с пустыми репозиториями;
  • автоматическое создание отсутствующих веток;
  • логирование операций и ошибок через safe_log.

Classes: GitManager: Менеджер для работы с Git-репозиториями.

Dependencies:

  • GitPython
  • pathlib
  • shutil
  • urllib.parse

Notes:

  • Для работы используется библиотека GitPython.
  • Все операции выполняются синхронно.
  • При возникновении ошибок методы логируют исключения и возвращают безопасные значения вместо проброса ошибок.
  • Временные локальные директории могут автоматически удаляться при повторной инициализации менеджера.

Example:

manager = GitManager( ... repo_url="https://github.com/user/repo.git", ... branch="main", ... username="user", ... password="token", ... )

manager.clone() True

manager.add() (True, True)

manager.commit("Update configs") True

manager.push() True

GitManager Objects

class GitManager()

Класс для управления Git репозиториями (клон, пуш, коммит).

Обеспечивает потокобезопасные операции с Git репозиториями, включая клонирование, добавление файлов, коммит, push, pull и очистку. Поддерживает аутентификацию по логину/паролю.

Attributes:

  • repo_url - URL удалённого репозитория
  • branch - Ветка по умолчанию для операций (None = ветка по умолчанию)
  • username - Имя пользователя для аутентификации
  • password - Пароль или токен для аутентификации
  • destination_path - Путь к локальной копии репозитория
  • auth_url - URL с встроенными данными аутентификации
  • repo - Объект Git репозитория (None после инициализации, создаётся при clone())
  • Repo - Класс GitPython Repo
  • GitCommandError - Класс исключения GitPython

__init__

def __init__(repo_url: str,
             branch: str | None = None,
             username: str | None = None,
             password: str | None = None,
             destination_path: str | None = None) -> None

Инициализация менеджера репозитория.

Arguments:

  • repo_url - URL репозитория (поддерживаются https:// и git://)
  • branch - Ветка по умолчанию для операций клонирования и пуша
  • username - Имя пользователя для HTTPS аутентификации
  • password - Пароль или персональный токен доступа
  • destination_path - Путь для клонирования (None = создаётся из имени репозитория)

Raises:

  • Exception - При ошибке удаления существующей директории

clone

def clone(branch: str | None = None) -> bool

Клонирует репозиторий в указанную директорию.

Если указанная ветка не существует в удалённом репозитории, создаёт её автоматически. Поддерживает пустые репозитории (без коммитов).

Arguments:

  • branch - Конкретная ветка для клонирования (переопределяет branch из init)

Returns:

True если клонирование успешно, False в противном случае

add

def add(files: list[str] | None = None) -> tuple[bool, bool]

Добавляет файлы в индекс Git.

Arguments:

  • files - Список файлов для добавления (None = все изменённые и неотслеживаемые файлы)

Returns:

Кортеж (успешность операции, были ли добавлены файлы):

  • - True, True - файлы успешно добавлены
  • - True, False - операция успешна, но нет файлов для добавления
  • - False, False - произошла ошибка

commit

def commit(message: str) -> bool

Создаёт коммит с проиндексированными изменениями.

Arguments:

  • message - Сообщение коммита

Returns:

True если коммит успешно создан, False в противном случае

push

def push(branch: str | None = None, force: bool = False) -> bool

Отправляет изменения в удалённый репозиторий.

Arguments:

  • branch - Ветка для пуша (None = текущая активная ветка)
  • force - Принудительный пуш (--force), перезаписывает удалённую ветку

Returns:

True если пуш успешен, False в противном случае

pull

def pull() -> bool

Выполняет pull (fetch + merge) последних изменений из удалённого репозитория.

Returns:

True если pull успешен, False в противном случае

cleanup

def cleanup() -> bool

Удаляет директорию с клонированным репозиторием.

Returns:

True если директория удалена или не существовала, False при ошибке

get_changed_hosts

def get_changed_hosts() -> list[str]

Возвращает список хостов, чьи конфигурации реально изменились.

Анализирует изменённые и неотслеживаемые файлы с расширением .rsc.

Returns:

Список имён хостов (без расширения .rsc), для которых есть изменения

get_status

def get_status() -> str | None

Возвращает статус репозитория (аналог git status).

Returns:

Строка с выводом git status или None, если репозиторий не загружен

is_dirty

def is_dirty() -> bool

Проверяет, есть ли незакоммиченные изменения в репозитории.

Учитывает как изменённые отслеживаемые файлы, так и неотслеживаемые.

Returns:

True если есть незакоммиченные изменения, False в противном случае

Модуль для отправки уведомлений через Telegram Bot API.

Содержит реализацию синхронного уведомителя TelegramNotifier, который используется для:

  • отправки текстовых уведомлений в Telegram;
  • получения и кэширования информации о чатах;
  • формирования итоговых отчетов о backup-процессах MikroTik;
  • логирования результатов отправки сообщений.

Основные возможности:

  • поддержка нескольких чатов;
  • автоматическое определение имен чатов;
  • форматирование сообщений с использованием HTML/Markdown;
  • отправка сводной статистики по успешным и ошибочным операциям;
  • обработка сетевых ошибок и логирование исключений.

Classes: TelegramNotifier: Сервис для отправки Telegram-уведомлений.

Dependencies:

  • niquests
  • datetime
  • TelegramConfig
  • safe_log

Example:

config = TelegramConfig(enabled=True, bot_token="TOKEN", chat_id="123456789") notifier = TelegramNotifier(config) notifier.send_message("Backup completed successfully")

FailedHost Objects

class FailedHost(TypedDict)

Информация об ошибке обработки хоста.

Attributes:

hostname: Имя хоста, на котором произошла ошибка.

error: Текст ошибки.

BackupResults Objects

class BackupResults(TypedDict)

Результаты backup-процесса.

Attributes:

successful: Список успешно обработанных хостов.

failed: Список ошибок обработки хостов.

TelegramNotifier Objects

class TelegramNotifier()

Синхронный сервис для отправки уведомлений через Telegram Bot API.

Класс отвечает за:

  • отправку текстовых сообщений в один или несколько Telegram-чатов;
  • получение и кэширование информации о чатах;
  • формирование итоговых уведомлений о выполнении backup-задач;
  • логирование успешных и ошибочных операций отправки.

Использует Telegram Bot API через HTTP-запросы библиотеки niquests.

Attributes:

config (TelegramConfig): Конфигурация Telegram-уведомлений.

base_url (str): Базовый URL Telegram Bot API для текущего бота.

chat_cache (dict[str, str]): Кэш отображаемых имен чатов по их ID.

Example:

config = TelegramConfig(enabled=True, bot_token="TOKEN", chat_id="123456") notifier = TelegramNotifier(config) notifier.send_message("Backup completed")

Notes:

  • Поддерживаются личные чаты, группы, супергруппы и каналы.
  • Имена чатов автоматически загружаются через метод getChat.
  • При ошибках сетевого взаимодействия исключения не пробрасываются наружу, а логируются через safe_log.

__init__

def __init__(config: TelegramConfig) -> None

Инициализирует экземпляр TelegramNotifier.

Создает базовый URL для Telegram Bot API, инициализирует кэш имен чатов и при необходимости заранее загружает информацию об основном и дополнительных чатах.

Arguments:

config (TelegramConfig): Конфигурация Telegram-уведомлений.

Attributes:

config (TelegramConfig): Сохраненная конфигурация уведомителя.

base_url (str): URL Telegram Bot API для текущего бота.

chat_cache (dict[str, str]): Кэш имен чатов, где ключ — ID чата.

Notes:

Предзагрузка информации о чатах выполняется только если уведомления включены (config.enabled == True).

get_chat_name

def get_chat_name(chat_id: str) -> str

Возвращает отображаемое имя Telegram-чата.

Если имя отсутствует в кэше, выполняется попытка загрузить информацию о чате через Telegram API.

Arguments:

chat_id (str): Идентификатор Telegram-чата.

Returns:

str: Имя чата или fallback-значение вида Chat <id>.

send_message

def send_message(message: str, chat_id: str | None = None) -> bool

Отправляет текстовое сообщение в Telegram.

Сообщение может быть отправлено:

  • в указанный чат;
  • в основной чат из конфигурации;
  • во все дополнительные чаты из конфигурации.

Arguments:

message (str): Текст сообщения.

chat_id (str | None, optional): ID целевого чата. Если не указан, используется основной чат из конфигурации.

Returns:

bool: True, если сообщение успешно отправлено во все целевые чаты, иначе False.

Notes:

  • Используется метод Telegram Bot API sendMessage.
  • Форматирование сообщения определяется параметром config.parse_mode.
  • Ошибки отправки и исключения логируются через safe_log.

send_complete_notification

def send_complete_notification(results: BackupResults,
                               changed_hosts: list[str]) -> None

Отправляет итоговое уведомление о выполнении backup-процесса.

Формирует единое Telegram-сообщение со статистикой:

  • успешных хостов;
  • хостов с изменениями;
  • ошибок выполнения;
  • времени завершения операции.

Arguments:

results (BackupResults): Структурированный результат обработки хостов.

changed_hosts (list[str]): Список хостов, в которых обнаружены изменения.

Notes:

Если Telegram-уведомления отключены, метод завершает работу без отправки сообщения.

MikroTik Backup - скрипт для параллельного сбора конфигураций с устройств MikroTik и сохранения их в Git репозиторий.

Этот модуль предназначен для автоматизации резервного копирования конфигураций сетевых устройств MikroTik RouterOS через SSH. Скрипт подключается к списку хостов, выполняет команду /export (с опцией show-sensitive или без), сохраняет полученные конфигурации в локальный Git репозиторий и отправляет изменения в удалённый репозиторий. Также поддерживает отправку уведомлений в Telegram.

Основные возможности: - Параллельное подключение к множеству устройств MikroTik через SSH - Аутентификация по SSH-ключам с проверкой прав доступа к ключам - Поддержка чувствительных данных (флаг sensitive для экспорта паролей) - Потокобезопасное логирование с блокировками - Автоматическое клонирование Git репозитория перед началом работы - Сохранение конфигураций в файлы с именем {hostname}.rsc - Определение изменений в конфигурациях через Git diff - Автоматический коммит и push изменений в удалённый Git репозиторий - Отправка уведомлений в Telegram о статусе выполнения - Очистка временных файлов после завершения работы

get_key_path

def get_key_path(key_path: str) -> Path

Преобразует строку в путь и проверяет существование ключа.

Arguments:

  • key_path - Путь к файлу ключа (абсолютный или относительный)

Returns:

Абсолютный путь к файлу ключа

Raises:

  • FileNotFoundError - Если файл ключа не существует
  • ValueError - Если путь указывает на директорию, а не файл
  • PermissionError - Если права доступа к ключу небезопасны (слишком открыты)

get_backup

def get_backup(params: GetBackupConfig) -> str

Получает конфигурацию с MikroTik через SSH.

Arguments:

  • params - Параметры подключения (хост, порт, пользователь, ключ, флаги)

Returns:

Конфигурация RouterOS в виде текста

Raises:

  • ConnectionError - При проблемах с подключением
  • AuthenticationException - При ошибке аутентификации (paramiko)
  • SSHException - При других SSH ошибках (paramiko)
  • TimeoutError - При превышении таймаута
  • ValueError - При получении пустой или слишком короткой конфигурации

write_file

def write_file(filename: str, content: str) -> None

Записывает содержимое в файл с проверкой успешности записи.

Arguments:

  • filename - Имя или путь к файлу для записи
  • content - Содержимое для записи в файл

Raises:

  • OSError - Если записано 0 байт или произошла ошибка при записи
  • IOError - При других ошибках ввода/вывода

Notes:

Файл открывается в режиме записи с кодировкой UTF-8. После записи выполняется принудительный сброс буфера (flush) и синхронизация (fsync).

load_config

def load_config(config_file: str) -> dict[str, Any]

Загружает конфигурацию из YAML файла.

Arguments:

  • config_file - Путь к YAML файлу конфигурации

Returns:

Словарь с данными конфигурации (ключи - строки, значения - любые типы)

parse_telegram_config

def parse_telegram_config(config: dict[str, Any]) -> TelegramConfig

Парсит конфигурацию Telegram из словаря.

Arguments:

  • config - Словарь с конфигурацией

Returns:

Объект TelegramConfig

process_host

def process_host(host_config: dict[str, Any], git_path: str,
                 defaults: dict[str, Any]) -> tuple[str, bool, str | None]

Обрабатывает один хост: получает бэкап и сохраняет файл.

Arguments:

  • host_config - Конфигурация хоста (содержит hostname, username, port, timeout, sensitive, key_file)
  • git_path - Путь к директории git для сохранения файлов
  • defaults - Словарь с настройками по умолчанию (key_file, username, port, timeout, sensitive)

Returns:

Кортеж (hostname, success, error_message):

  • hostname: Имя хоста
  • success: True если успешно, False если ошибка
  • error_message: Текст ошибки или None при успехе

process_hosts_parallel

def process_hosts_parallel(config: dict[str, Any],
                           defaults: dict[str, Any]) -> dict[str, list]

Параллельная обработка хостов с использованием ThreadPoolExecutor.

Arguments:

  • config - Основная конфигурация, содержащая:
    • hosts: список конфигураций хостов
    • git.path: путь к git директории
    • parallel.max_workers: количество параллельных потоков (по умолчанию 5)
    • parallel.timeout_per_host: таймаут на один хост в секундах (по умолчанию 60)
  • defaults - Словарь с настройками по умолчанию для хостов

Returns:

Словарь с двумя ключами:

  • "successful": список успешно обработанных хостов (list[str])
  • "failed": список хостов с ошибками (list[dict]), каждый dict содержит "hostname" и "error"

print_statistics

def print_statistics(results: dict[str, list[Any]]) -> None

Выводит статистику обработки хостов.

Arguments:

  • results - Словарь с результатами обработки, содержащий:
    • "successful": список успешно обработанных хостов (list[str])
    • "failed": список словарей с ошибками, каждый словарь содержит "hostname" (str) и "error" (str)

commit_changes

def commit_changes(git_manager: GitManager, results: dict[str, list[Any]],
                   commit_message_template: str) -> list[str]

Коммитит изменения в Git.

Arguments:

  • git_manager - Менеджер Git репозитория (должен быть уже клонирован)
  • results - Словарь с результатами обработки, содержащий ключ "successful" (список успешных хостов)
  • commit_message_template - Шаблон сообщения коммита (к нему через пробел добавится список хостов)

Returns:

Список хостов, в которых были обнаружены изменения

parse_email_config

def parse_email_config(config: dict[str, Any]) -> EmailConfig

Парсит email конфигурацию.

main

def main() -> None

Главная функция с поддержкой Telegram уведомлений.

About

No description, website, or topics provided.

Resources

Readme
MIT license
Activity

Stars

0 stars

Watchers

0 watching

Forks

1 fork
Report repository

Releases

Packages

Contributors

Languages