Главная Проекты Статьи 🔑 Генератор паролей ⚡ VLESS Конвертер Контакты
← Все статьи
MikroTik

Автоматизация MikroTik через Python: RouterOS API

RouterOS API

MikroTik предоставляет встроенный API (порт 8728, SSL — 8729), через который можно управлять роутером программно: читать и изменять конфигурацию, получать статистику, выполнять команды — всё то же самое, что доступно в терминале или Winbox.

Это позволяет автоматизировать рутинные задачи: обновление списков блокировок, резервное копирование, мониторинг с алертами в Telegram, массовое управление несколькими роутерами.

Требования

  • Python 3.8+
  • Библиотека routeros-api
  • API включён на роутере (по умолчанию включён)

Шаг 1 — Включить API на MikroTik

/ip/service/enable api
/ip/service/set api port=8728

Для безопасности ограничьте доступ к API только с нужного IP:

/ip/service/set api address=192.168.1.0/24

Создайте отдельного пользователя с ограниченными правами для скриптов:

/user/group/add name=api-readonly policy=read,api,test
/user/add name=apiuser group=api-readonly password=StrongApiPass123

Шаг 2 — Установка библиотеки

pip install routeros-api

Шаг 3 — Подключение к роутеру

import routeros_api

connection = routeros_api.RouterOsApiPool(
    host='192.168.1.1',
    username='apiuser',
    password='StrongApiPass123',
    port=8728,
    plaintext_login=True,  # для RouterOS 6.49+
)
api = connection.get_api()
print('Connected to MikroTik')
connection.disconnect()

Шаг 4 — Чтение данных

Список интерфейсов с трафиком

import routeros_api

def get_interfaces(host, user, password):
    pool = routeros_api.RouterOsApiPool(host, username=user, password=password, plaintext_login=True)
    api = pool.get_api()

    ifaces = api.get_resource('/interface')
    for iface in ifaces.get():
        tx = int(iface.get('tx-byte', 0))
        rx = int(iface.get('rx-byte', 0))
        print(f"{iface['name']:20s}  TX: {tx//1024//1024:6d} MB  RX: {rx//1024//1024:6d} MB")

    pool.disconnect()

get_interfaces('192.168.1.1', 'apiuser', 'StrongApiPass123')

Активные DHCP-аренды (кто в сети)

leases = api.get_resource('/ip/dhcp-server/lease')
for lease in leases.get():
    if lease.get('status') == 'bound':
        print(f"{lease.get('host-name', '?'):20s}  {lease['address']:15s}  {lease['mac-address']}")

Загрузка CPU и свободная память

res = api.get_resource('/system/resource')
info = res.get()[0]
print(f"CPU load:    {info['cpu-load']}%")
print(f"Free memory: {int(info['free-memory']) // 1024} KB")
print(f"Uptime:      {info['uptime']}")

Шаг 5 — Управление Address Lists

Автоматическое обновление списка IP-адресов (например, для маршрутизации через VLESS):

import routeros_api
import requests

MIKROTIK_HOST = '192.168.1.1'
MIKROTIK_USER = 'apiuser'
MIKROTIK_PASS = 'StrongApiPass123'
LIST_NAME = 'proxy-list'

# Загружаем актуальный список IP с внешнего источника
def fetch_blocked_ips():
    url = 'https://raw.githubusercontent.com/zapret-info/z-i/master/dump.csv'
    resp = requests.get(url, timeout=30)
    ips = set()
    for line in resp.text.splitlines():
        parts = line.split(';')
        if parts:
            for ip in parts[0].split('|'):
                ip = ip.strip()
                if ip and not ip.startswith('#'):
                    ips.add(ip)
    return ips

def update_address_list(ips):
    pool = routeros_api.RouterOsApiPool(
        MIKROTIK_HOST, username=MIKROTIK_USER,
        password=MIKROTIK_PASS, plaintext_login=True
    )
    api = pool.get_api()
    addr_list = api.get_resource('/ip/firewall/address-list')

    # Удаляем старые записи списка
    old = addr_list.get(list=LIST_NAME)
    for entry in old:
        addr_list.remove(id=entry['id'])
    print(f'Removed {len(old)} old entries')

    # Добавляем новые
    added = 0
    for ip in ips:
        try:
            addr_list.add(list=LIST_NAME, address=ip, comment='auto-updated')
            added += 1
        except Exception:
            pass  # пропускаем некорректные адреса

    print(f'Added {added} entries to {LIST_NAME}')
    pool.disconnect()

ips = fetch_blocked_ips()
update_address_list(ips)

Шаг 6 — Резервное копирование конфигурации

import routeros_api
import paramiko
import datetime
import os

def backup_mikrotik(host, user, password, ssh_user, ssh_pass, backup_dir='/backups'):
    # Создаём бэкап через API
    pool = routeros_api.RouterOsApiPool(host, username=user, password=password, plaintext_login=True)
    api = pool.get_api()

    date_str = datetime.datetime.now().strftime('%Y%m%d_%H%M')
    backup_name = f'backup_{date_str}'

    system = api.get_resource('/system/backup')
    system.call('save', {'name': backup_name})
    print(f'Backup created: {backup_name}.backup')
    pool.disconnect()

    # Скачиваем через SFTP
    os.makedirs(backup_dir, exist_ok=True)
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(host, username=ssh_user, password=ssh_pass)
    sftp = ssh.open_sftp()
    remote_path = f'{backup_name}.backup'
    local_path = os.path.join(backup_dir, remote_path)
    sftp.get(remote_path, local_path)
    sftp.remove(remote_path)  # удаляем с роутера
    sftp.close()
    ssh.close()
    print(f'Downloaded to {local_path}')

backup_mikrotik(
    host='192.168.1.1',
    user='apiuser', password='StrongApiPass123',
    ssh_user='admin', ssh_pass='AdminPass',
    backup_dir='/opt/mikrotik-backups'
)

Шаг 7 — Мониторинг с алертами в Telegram

import routeros_api
import requests
import time

TELEGRAM_TOKEN = 'YOUR_BOT_TOKEN'
TELEGRAM_CHAT_ID = 'YOUR_CHAT_ID'
MIKROTIK_HOST = '192.168.1.1'
WAN_INTERFACE = 'ether1'
CPU_THRESHOLD = 80  # %

def send_telegram(message):
    url = f'https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage'
    requests.post(url, json={'chat_id': TELEGRAM_CHAT_ID, 'text': message, 'parse_mode': 'HTML'})

def check_mikrotik():
    try:
        pool = routeros_api.RouterOsApiPool(
            MIKROTIK_HOST, username='apiuser',
            password='StrongApiPass123', plaintext_login=True
        )
        api = pool.get_api()

        # Проверяем CPU
        res = api.get_resource('/system/resource').get()[0]
        cpu = int(res['cpu-load'])
        if cpu > CPU_THRESHOLD:
            send_telegram(f'⚠️ MikroTik: высокая загрузка CPU — {cpu}%')

        # Проверяем WAN-интерфейс
        ifaces = api.get_resource('/interface').get()
        for iface in ifaces:
            if iface['name'] == WAN_INTERFACE and iface.get('running') != 'true':
                send_telegram(f'🔴 MikroTik: интерфейс {WAN_INTERFACE} не работает!')

        pool.disconnect()
    except Exception as e:
        send_telegram(f'🔴 MikroTik недоступен!
{e}')

# Запуск каждые 60 секунд
while True:
    check_mikrotik()
    time.sleep(60)

Шаг 8 — Управление несколькими роутерами

import routeros_api
from concurrent.futures import ThreadPoolExecutor

ROUTERS = [
    {'host': '192.168.1.1',   'name': 'Main'},
    {'host': '192.168.10.1',  'name': 'Office'},
    {'host': '192.168.20.1',  'name': 'Remote'},
]

def get_router_info(router):
    try:
        pool = routeros_api.RouterOsApiPool(
            router['host'], username='apiuser',
            password='StrongApiPass123', plaintext_login=True,
        )
        api = pool.get_api()
        res = api.get_resource('/system/resource').get()[0]
        identity = api.get_resource('/system/identity').get()[0]['name']
        pool.disconnect()
        return {
            'name': router['name'],
            'identity': identity,
            'cpu': res['cpu-load'],
            'uptime': res['uptime'],
            'version': res['version'],
        }
    except Exception as e:
        return {'name': router['name'], 'error': str(e)}

with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(get_router_info, ROUTERS))

for r in results:
    if 'error' in r:
        print(f"[{r['name']}] ERROR: {r['error']}")
    else:
        print(f"[{r['name']}] {r['identity']} | CPU: {r['cpu']}% | Up: {r['uptime']} | ROS: {r['version']}")

Запуск по расписанию (cron)

Добавьте задачи в crontab на сервере:

crontab -e
# Резервное копирование в 02:00 каждый день
0 2 * * * /usr/bin/python3 /opt/scripts/backup_mikrotik.py >> /var/log/mt-backup.log 2>&1

# Обновление списка блокировок в 03:00
0 3 * * * /usr/bin/python3 /opt/scripts/update_adlist.py >> /var/log/mt-adlist.log 2>&1

# Мониторинг (запускаем как systemd-сервис или screen)
# systemctl enable mt-monitor.service

Структура проекта

/opt/mikrotik-scripts/
├── config.py          # хосты, токены, пароли
├── backup.py          # резервное копирование
├── update_lists.py    # обновление address lists
├── monitor.py         # мониторинг с Telegram
├── inventory.py       # опрос нескольких роутеров
└── requirements.txt   # routeros-api paramiko requests

Устранение неполадок

Ошибка подключения

# Проверьте, что API включён:
# /ip/service/print — в строке api должен быть статус enabled
# Проверьте firewall:
# /ip/firewall/filter/print — убедитесь, что порт 8728 не заблокирован

Ошибка аутентификации на RouterOS 6

В RouterOS 6 используется другой метод логина. Уберите параметр plaintext_login:

pool = routeros_api.RouterOsApiPool(host, username=user, password=password)
# plaintext_login=True только для RouterOS 7+

Таймаут при большом количестве записей

pool = routeros_api.RouterOsApiPool(
    host, username=user, password=password,
    plaintext_login=True,
    timeout=30,  # увеличьте таймаут
)

Ссылки