# Автоматизация VyOS - REST API, Ansible и Python > Автоматизация VyOS через REST API, Ansible и Python - массовая конфигурация, CI/CD интеграция и мониторинг в Yandex Cloud и VK Cloud Source: https://opennix.org/docs/vyos/admin-guide/vyos-automation/ VyOS предоставляет мощные инструменты для автоматизации управления конфигурацией через REST API, GraphQL API, Ansible и Python. Автоматизация критически важна для управления парком устройств, обеспечения консистентности конфигурации и интеграции с системами DevOps. ## REST API VyOS предоставляет RESTful API для программного управления конфигурацией и получения операционной информации. ### Настройка REST API #### VyOS 1.5 (Circinus) ```bash configure # Создать API ключ set service https api keys id automation key 'your-api-key-here' # Включить REST API set service https api rest # Опционально: настроить HTTPS сертификат set service https certificates ca-certificate ca-vyos set service https certificates certificate vyos-cert # Настроить доступ set service https listen-address 192.168.1.1 commit save ``` #### VyOS 1.4 (Sagitta) ```bash configure # В версии 1.4 используется единый API endpoint set service https api keys id automation key 'your-api-key-here' commit save ``` ### Основные API Endpoints #### Configuration Management **Set Configuration** - применить конфигурационную команду: ```bash POST /configure Content-Type: application/json { "op": "set", "path": ["interfaces", "ethernet", "eth1", "address"], "value": "192.168.2.1/24", "key": "your-api-key-here" } ``` **Delete Configuration** - удалить конфигурационный элемент: ```bash POST /configure Content-Type: application/json { "op": "delete", "path": ["interfaces", "ethernet", "eth1", "address", "192.168.2.1/24"], "key": "your-api-key-here" } ``` **Show Configuration** - получить конфигурацию: ```bash POST /retrieve Content-Type: application/json { "op": "showConfig", "path": ["interfaces", "ethernet"], "key": "your-api-key-here" } ``` **Show Operational Data** - выполнить show команду: ```bash POST /show Content-Type: application/json { "op": "show", "path": ["interfaces"], "key": "your-api-key-here" } ``` #### Configuration File Operations **Save Configuration**: ```bash POST /config-file Content-Type: application/json { "op": "save", "key": "your-api-key-here" } ``` **Load Configuration from File**: ```bash POST /config-file Content-Type: application/json { "op": "load", "file": "/config/config.boot.backup", "key": "your-api-key-here" } ``` #### Image Management **Add System Image**: ```bash POST /image Content-Type: application/json { "op": "add", "url": "https://downloads.vyos.io/rolling/current/vyos-1.5-rolling-latest.iso", "key": "your-api-key-here" } ``` **Delete System Image**: ```bash POST /image Content-Type: application/json { "op": "delete", "name": "1.4.20230101", "key": "your-api-key-here" } ``` #### System Operations **Reboot**: ```bash POST /reboot Content-Type: application/json { "key": "your-api-key-here" } ``` **Power Off**: ```bash POST /poweroff Content-Type: application/json { "key": "your-api-key-here" } ``` ### Python REST API Client #### Базовый клиент ```python import requests import json from typing import Dict, List, Any, Optional class VyOSClient: """Клиент для работы с VyOS REST API""" def __init__(self, host: str, api_key: str, use_https: bool = True, verify_ssl: bool = True): self.host = host self.api_key = api_key self.protocol = "https" if use_https else "http" self.verify_ssl = verify_ssl self.base_url = f"{self.protocol}://{host}" def _make_request(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]: """Выполнить API запрос""" data['key'] = self.api_key url = f"{self.base_url}/{endpoint}" try: response = requests.post( url, json=data, verify=self.verify_ssl, timeout=30 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: raise Exception(f"API request failed: {e}") def set_config(self, path: List[str], value: Optional[str] = None) -> Dict[str, Any]: """Установить конфигурационное значение""" data = { "op": "set", "path": path } if value: data['value'] = value return self._make_request("configure", data) def delete_config(self, path: List[str]) -> Dict[str, Any]: """Удалить конфигурационный элемент""" data = { "op": "delete", "path": path } return self._make_request("configure", data) def show_config(self, path: List[str] = []) -> Dict[str, Any]: """Получить конфигурацию""" data = { "op": "showConfig", "path": path } return self._make_request("retrieve", data) def show(self, path: List[str]) -> Dict[str, Any]: """Выполнить show команду""" data = { "op": "show", "path": path } return self._make_request("show", data) def commit(self) -> Dict[str, Any]: """Применить конфигурацию""" data = {"op": "commit"} return self._make_request("configure", data) def save(self) -> Dict[str, Any]: """Сохранить конфигурацию""" data = {"op": "save"} return self._make_request("config-file", data) def batch_configure(self, commands: List[Dict[str, Any]]) -> Dict[str, Any]: """Выполнить пакетную конфигурацию""" results = [] for cmd in commands: if cmd['op'] == 'set': result = self.set_config(cmd['path'], cmd.get('value')) elif cmd['op'] == 'delete': result = self.delete_config(cmd['path']) results.append(result) # Применить изменения commit_result = self.commit() results.append(commit_result) # Сохранить save_result = self.save() results.append(save_result) return {"results": results} # Пример использования if __name__ == "__main__": # Создать клиент client = VyOSClient( host="192.168.1.1", api_key="your-api-key-here", verify_ssl=False # Для self-signed сертификатов ) # Настроить интерфейс client.set_config( path=["interfaces", "ethernet", "eth1", "address"], value="192.168.2.1/24" ) client.set_config( path=["interfaces", "ethernet", "eth1", "description"], value="LAN Interface" ) # Применить и сохранить client.commit() client.save() # Получить конфигурацию интерфейса config = client.show_config(path=["interfaces", "ethernet", "eth1"]) print(json.dumps(config, indent=2)) # Получить статус интерфейсов status = client.show(path=["interfaces"]) print(json.dumps(status, indent=2)) ``` #### Продвинутый клиент с контекстным менеджером ```python from contextlib import contextmanager import logging class VyOSConfigSession(VyOSClient): """Клиент с автоматическим commit/save и rollback при ошибках""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.logger = logging.getLogger(__name__) @contextmanager def config_session(self, auto_save: bool = True): """Контекстный менеджер для безопасной конфигурации""" try: yield self # При успехе - commit и save self.commit() if auto_save: self.save() self.logger.info("Configuration applied successfully") except Exception as e: # При ошибке - rollback self.logger.error(f"Configuration failed: {e}") try: self._make_request("configure", {"op": "discard"}) self.logger.info("Configuration rolled back") except: self.logger.error("Rollback failed") raise # Пример использования with VyOSConfigSession( host="192.168.1.1", api_key="your-api-key-here", verify_ssl=False ).config_session() as session: # Все изменения в этом блоке будут автоматически commit/save session.set_config( path=["interfaces", "ethernet", "eth1", "address"], value="192.168.2.1/24" ) session.set_config( path=["protocols", "static", "route", "10.0.0.0/8", "next-hop"], value="192.168.2.254" ) # При выходе из блока - автоматический commit и save # При ошибке - автоматический rollback ``` ## Ansible Automation Ansible предоставляет декларативный подход к управлению конфигурацией VyOS через модуль `vyos.vyos`. ### Установка Ansible Collection ```bash # Установить Ansible pip install ansible # Установить VyOS collection ansible-galaxy collection install vyos.vyos ``` ### Inventory Configuration #### inventory/hosts.yml ```yaml all: children: vyos_routers: hosts: border-router: ansible_host: 192.168.1.1 ansible_network_os: vyos.vyos.vyos ansible_connection: ansible.netcommon.network_cli ansible_user: automation ansible_password: "{{ vault_vyos_password }}" branch-router: ansible_host: 10.1.1.1 ansible_network_os: vyos.vyos.vyos ansible_connection: ansible.netcommon.network_cli ansible_user: automation ansible_password: "{{ vault_vyos_password }}" vars: ansible_python_interpreter: /usr/bin/python3 ``` #### Хранение паролей (Ansible Vault) ```bash # Создать зашифрованный файл с паролями ansible-vault create inventory/group_vars/vyos_routers/vault.yml # Содержимое vault.yml vault_vyos_password: "secure-password-here" ``` ### Базовые Playbooks #### Конфигурация интерфейса ```yaml --- # playbooks/configure_interface.yml - name: Configure Ethernet Interface hosts: vyos_routers gather_facts: false vars: interface_name: eth1 interface_address: 192.168.2.1/24 interface_description: "LAN Interface" tasks: - name: Configure interface address vyos.vyos.vyos_config: lines: - set interfaces ethernet {{ interface_name }} address {{ interface_address }} - set interfaces ethernet {{ interface_name }} description '{{ interface_description }}' save: true ``` #### Конфигурация BGP ```yaml --- # playbooks/configure_bgp.yml - name: Configure BGP hosts: border-router gather_facts: false vars: local_asn: 65001 router_id: 192.168.1.1 neighbors: - ip: 10.0.0.2 remote_asn: 65002 description: "ISP1" - ip: 10.0.1.2 remote_asn: 65003 description: "ISP2" tasks: - name: Configure BGP basic settings vyos.vyos.vyos_config: lines: - set protocols bgp {{ local_asn }} parameters router-id {{ router_id }} save: false - name: Configure BGP neighbors vyos.vyos.vyos_config: lines: - set protocols bgp {{ local_asn }} neighbor {{ item.ip }} remote-as {{ item.remote_asn }} - set protocols bgp {{ local_asn }} neighbor {{ item.ip }} description '{{ item.description }}' - set protocols bgp {{ local_asn }} neighbor {{ item.ip }} address-family ipv4-unicast save: false loop: "{{ neighbors }}" - name: Commit and save configuration vyos.vyos.vyos_config: save: true ``` #### VPN Site-to-Site ```yaml --- # playbooks/configure_vpn.yml - name: Configure IPsec Site-to-Site VPN hosts: vyos_routers gather_facts: false vars: local_peer: 203.0.113.1 remote_peer: 203.0.113.2 psk: "{{ vault_ipsec_psk }}" local_network: 192.168.1.0/24 remote_network: 192.168.2.0/24 tasks: - name: Configure IPsec authentication vyos.vyos.vyos_config: lines: - set vpn ipsec ike-group IKE-SITE authentication mode pre-shared-secret - set vpn ipsec ike-group IKE-SITE proposal 1 dh-group 14 - set vpn ipsec ike-group IKE-SITE proposal 1 encryption aes256 - set vpn ipsec ike-group IKE-SITE proposal 1 hash sha256 - set vpn ipsec esp-group ESP-SITE proposal 1 encryption aes256 - set vpn ipsec esp-group ESP-SITE proposal 1 hash sha256 save: false - name: Configure IPsec site-to-site peer vyos.vyos.vyos_config: lines: - set vpn ipsec site-to-site peer {{ remote_peer }} authentication mode pre-shared-secret - set vpn ipsec site-to-site peer {{ remote_peer }} authentication pre-shared-secret '{{ psk }}' - set vpn ipsec site-to-site peer {{ remote_peer }} ike-group IKE-SITE - set vpn ipsec site-to-site peer {{ remote_peer }} local-address {{ local_peer }} - set vpn ipsec site-to-site peer {{ remote_peer }} tunnel 1 esp-group ESP-SITE - set vpn ipsec site-to-site peer {{ remote_peer }} tunnel 1 local prefix {{ local_network }} - set vpn ipsec site-to-site peer {{ remote_peer }} tunnel 1 remote prefix {{ remote_network }} save: true ``` ### Продвинутые Playbooks #### Массовое развертывание VLAN ```yaml --- # playbooks/deploy_vlans.yml - name: Deploy VLANs across fleet hosts: vyos_routers gather_facts: false vars: vlans: - id: 10 description: "Management" address: "192.168.10.1/24" dhcp_start: "192.168.10.100" dhcp_stop: "192.168.10.200" - id: 20 description: "Servers" address: "192.168.20.1/24" dhcp_start: "192.168.20.100" dhcp_stop: "192.168.20.200" - id: 30 description: "Workstations" address: "192.168.30.1/24" dhcp_start: "192.168.30.100" dhcp_stop: "192.168.30.250" tasks: - name: Create VLAN interfaces vyos.vyos.vyos_config: lines: - set interfaces ethernet eth0 vif {{ item.id }} description '{{ item.description }}' - set interfaces ethernet eth0 vif {{ item.id }} address {{ item.address }} save: false loop: "{{ vlans }}" - name: Configure DHCP for VLANs vyos.vyos.vyos_config: lines: - set service dhcp-server shared-network-name VLAN{{ item.id }} subnet {{ item.address | ansible.netcommon.ipaddr('network/prefix') }} default-router {{ item.address | ansible.netcommon.ipaddr('address') }} - set service dhcp-server shared-network-name VLAN{{ item.id }} subnet {{ item.address | ansible.netcommon.ipaddr('network/prefix') }} range 0 start {{ item.dhcp_start }} - set service dhcp-server shared-network-name VLAN{{ item.id }} subnet {{ item.address | ansible.netcommon.ipaddr('network/prefix') }} range 0 stop {{ item.dhcp_stop }} - set service dhcp-server shared-network-name VLAN{{ item.id }} subnet {{ item.address | ansible.netcommon.ipaddr('network/prefix') }} name-server 8.8.8.8 save: false loop: "{{ vlans }}" - name: Commit configuration vyos.vyos.vyos_config: save: true ``` #### Резервное копирование конфигурации ```yaml --- # playbooks/backup_configs.yml - name: Backup VyOS configurations hosts: vyos_routers gather_facts: true vars: backup_dir: "./backups" tasks: - name: Create backup directory local_action: module: file path: "{{ backup_dir }}/{{ inventory_hostname }}" state: directory run_once: true - name: Get running configuration vyos.vyos.vyos_command: commands: - show configuration commands register: config_output - name: Save configuration to file local_action: module: copy content: "{{ config_output.stdout[0] }}" dest: "{{ backup_dir }}/{{ inventory_hostname }}/config-{{ ansible_date_time.iso8601_basic_short }}.txt" - name: Get system information vyos.vyos.vyos_command: commands: - show version - show interfaces - show ip route register: system_info - name: Save system information local_action: module: copy content: "{{ system_info.stdout | join('\n\n=====\n\n') }}" dest: "{{ backup_dir }}/{{ inventory_hostname }}/system-info-{{ ansible_date_time.iso8601_basic_short }}.txt" ``` ### Ansible Roles #### Структура роли ``` roles/ └── vyos-base/ ├── defaults/ │ └── main.yml ├── tasks/ │ ├── main.yml │ ├── interfaces.yml │ ├── services.yml │ └── security.yml ├── templates/ │ ├── firewall.j2 │ └── nat.j2 └── vars/ └── main.yml ``` #### roles/vyos-base/defaults/main.yml ```yaml --- # Default variables vyos_timezone: Europe/Moscow vyos_ntp_servers: - 0.ru.pool.ntp.org - 1.ru.pool.ntp.org vyos_ssh_port: 22 vyos_ssh_enable_password_auth: false vyos_enable_firewall: true vyos_enable_nat: true ``` #### roles/vyos-base/tasks/main.yml ```yaml --- - name: Configure system settings include_tasks: system.yml - name: Configure interfaces include_tasks: interfaces.yml - name: Configure services include_tasks: services.yml - name: Configure security include_tasks: security.yml when: vyos_enable_firewall ``` #### Использование роли ```yaml --- # playbooks/deploy_base_config.yml - name: Deploy base VyOS configuration hosts: vyos_routers gather_facts: false roles: - role: vyos-base vars: vyos_timezone: Europe/Moscow vyos_enable_firewall: true ``` ## Configuration Management ### Git-based Configuration Management #### Структура репозитория ``` vyos-configs/ ├── devices/ │ ├── border-router/ │ │ └── config.boot │ ├── branch-router-01/ │ │ └── config.boot │ └── branch-router-02/ │ └── config.boot ├── templates/ │ ├── base.boot.j2 │ ├── branch.boot.j2 │ └── border.boot.j2 ├── scripts/ │ ├── deploy.py │ ├── validate.py │ └── backup.py └── README.md ``` #### Скрипт автоматического развертывания ```python #!/usr/bin/env python3 # scripts/deploy.py import sys import argparse from pathlib import Path import subprocess from vyos_client import VyOSClient # Из примера выше def deploy_config(device: str, config_file: Path, api_key: str): """Развернуть конфигурацию на устройство""" # Читать конфигурацию with open(config_file) as f: config_lines = f.readlines() # Создать клиент client = VyOSClient(host=device, api_key=api_key, verify_ssl=False) # Парсить и применить конфигурацию for line in config_lines: line = line.strip() if not line or line.startswith('#'): continue if line.startswith('set '): # Извлечь path и value из команды set parts = line.split()[1:] # Убрать 'set' # Найти value (всё после последнего пробела, если есть кавычки) if "'" in line: path_parts = [] value = None in_quotes = False current = "" for part in parts: if part.startswith("'"): in_quotes = True current = part[1:] elif part.endswith("'"): current += " " + part[:-1] value = current in_quotes = False elif in_quotes: current += " " + part else: path_parts.append(part) client.set_config(path=path_parts, value=value) else: client.set_config(path=parts) # Commit и save client.commit() client.save() print(f"Configuration deployed to {device}") def main(): parser = argparse.ArgumentParser(description='Deploy VyOS configuration') parser.add_argument('device', help='Device hostname or IP') parser.add_argument('config', type=Path, help='Configuration file') parser.add_argument('--api-key', required=True, help='API key') args = parser.parse_args() if not args.config.exists(): print(f"Error: Configuration file {args.config} not found") sys.exit(1) deploy_config(args.device, args.config, args.api_key) if __name__ == '__main__': main() ``` ### CI/CD Integration #### GitLab CI Example ```yaml # .gitlab-ci.yml stages: - validate - test - deploy variables: VYOS_API_KEY: ${CI_VYOS_API_KEY} validate_syntax: stage: validate image: python:3.11 script: - pip install -r requirements.txt - python scripts/validate.py devices/ test_staging: stage: test image: python:3.11 script: - python scripts/deploy.py staging-router devices/border-router/config.boot --api-key ${VYOS_API_KEY} - python scripts/test_connectivity.py staging-router only: - merge_requests deploy_production: stage: deploy image: python:3.11 script: - | for device in devices/*/; do device_name=$(basename $device) echo "Deploying to $device_name" python scripts/deploy.py $device_name $device/config.boot --api-key ${VYOS_API_KEY} done only: - main when: manual ``` ## Практические сценарии ### Сценарий 1: Массовая конфигурация интерфейсов через API ```python #!/usr/bin/env python3 """ Массовая конфигурация интерфейсов на множестве устройств """ from vyos_client import VyOSClient from concurrent.futures import ThreadPoolExecutor, as_completed import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Список устройств DEVICES = [ {"host": "192.168.1.1", "name": "border-router"}, {"host": "192.168.2.1", "name": "branch-01"}, {"host": "192.168.3.1", "name": "branch-02"}, {"host": "192.168.4.1", "name": "branch-03"}, ] # Конфигурация интерфейсов INTERFACE_CONFIG = { "eth0": { "description": "WAN", "address": "dhcp" }, "eth1": { "description": "LAN", "address": None, # Устанавливается индивидуально "mtu": "1500" } } API_KEY = "your-api-key-here" def configure_device(device: dict) -> dict: """Сконфигурировать одно устройство""" try: client = VyOSClient( host=device['host'], api_key=API_KEY, verify_ssl=False ) logger.info(f"Configuring {device['name']} ({device['host']})") # Конфигурировать интерфейсы for iface, config in INTERFACE_CONFIG.items(): # Описание if 'description' in config: client.set_config( path=["interfaces", "ethernet", iface, "description"], value=config['description'] ) # Адрес (если не dhcp) if config.get('address') == 'dhcp': client.set_config( path=["interfaces", "ethernet", iface, "address"], value="dhcp" ) elif config.get('address'): client.set_config( path=["interfaces", "ethernet", iface, "address"], value=config['address'] ) # MTU if 'mtu' in config: client.set_config( path=["interfaces", "ethernet", iface, "mtu"], value=config['mtu'] ) # Commit и save client.commit() client.save() logger.info(f"Successfully configured {device['name']}") return {"device": device['name'], "status": "success"} except Exception as e: logger.error(f"Failed to configure {device['name']}: {e}") return {"device": device['name'], "status": "failed", "error": str(e)} def main(): """Главная функция""" results = [] # Параллельная конфигурация устройств with ThreadPoolExecutor(max_workers=5) as executor: futures = { executor.submit(configure_device, device): device for device in DEVICES } for future in as_completed(futures): result = future.result() results.append(result) # Вывод результатов print("\n=== Configuration Summary ===") success_count = sum(1 for r in results if r['status'] == 'success') print(f"Total devices: {len(results)}") print(f"Successful: {success_count}") print(f"Failed: {len(results) - success_count}") for result in results: status_icon = "✓" if result['status'] == 'success' else "✗" print(f"{status_icon} {result['device']}: {result['status']}") if 'error' in result: print(f" Error: {result['error']}") if __name__ == "__main__": main() ``` ### Сценарий 2: Ansible Playbook для управления парком роутеров ```yaml --- # playbooks/fleet_management.yml - name: VyOS Fleet Management Playbook hosts: vyos_routers gather_facts: false vars: base_timezone: Europe/Moscow base_ntp_servers: - 0.ru.pool.ntp.org - 1.ru.pool.ntp.org - ntp1.yandex.ru base_dns_servers: - 8.8.8.8 - 8.8.4.4 - 77.88.8.8 ssh_hardening: port: 22 disable_password: true disable_root: true tasks: - name: Configure system time vyos.vyos.vyos_config: lines: - set system time-zone {{ base_timezone }} save: false tags: [system, time] - name: Configure NTP vyos.vyos.vyos_config: lines: - set service ntp server {{ item }} save: false loop: "{{ base_ntp_servers }}" tags: [system, ntp] - name: Configure DNS forwarding vyos.vyos.vyos_config: lines: - set service dns forwarding cache-size 10000 - set service dns forwarding listen-address {{ ansible_host }} - set service dns forwarding name-server {{ item }} save: false loop: "{{ base_dns_servers }}" tags: [services, dns] - name: Harden SSH vyos.vyos.vyos_config: lines: - set service ssh port {{ ssh_hardening.port }} - delete service ssh disable-password-authentication - set service ssh disable-password-authentication save: false when: ssh_hardening.disable_password tags: [security, ssh] - name: Configure logging vyos.vyos.vyos_config: lines: - set system syslog global facility all level info - set system syslog host 192.168.1.10 facility all level warning save: false tags: [system, logging] - name: Commit configuration vyos.vyos.vyos_config: save: true tags: [always] - name: Verify configuration vyos.vyos.vyos_command: commands: - show configuration - show service ntp - show service dns forwarding statistics register: verify_output tags: [verify] - name: Display verification results debug: var: verify_output.stdout_lines tags: [verify] ``` ### Сценарий 3: Автоматизация резервного копирования с ротацией ```python #!/usr/bin/env python3 """ Автоматическое резервное копирование конфигурации VyOS с ротацией """ import os import sys from datetime import datetime, timedelta from pathlib import Path import gzip import shutil from vyos_client import VyOSClient # Конфигурация BACKUP_DIR = Path("/opt/backups/vyos") RETENTION_DAYS = 30 COMPRESSION_AFTER_DAYS = 7 DEVICES = [ {"host": "192.168.1.1", "name": "border-router"}, {"host": "192.168.2.1", "name": "branch-01"}, {"host": "192.168.3.1", "name": "branch-02"}, ] API_KEY = os.environ.get("VYOS_API_KEY", "your-api-key-here") def backup_device(device: dict, backup_dir: Path) -> bool: """Создать резервную копию конфигурации устройства""" try: client = VyOSClient( host=device['host'], api_key=API_KEY, verify_ssl=False ) # Получить конфигурацию config_response = client.show_config(path=[]) if not config_response.get('success'): print(f"Error getting config from {device['name']}") return False config_data = config_response.get('data', '') # Создать директорию для устройства device_backup_dir = backup_dir / device['name'] device_backup_dir.mkdir(parents=True, exist_ok=True) # Имя файла с timestamp timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") backup_file = device_backup_dir / f"config-{timestamp}.boot" # Сохранить конфигурацию with open(backup_file, 'w') as f: f.write(config_data) print(f"Backed up {device['name']} to {backup_file}") # Получить дополнительную информацию version_response = client.show(path=["version"]) if version_response.get('success'): version_file = device_backup_dir / f"version-{timestamp}.txt" with open(version_file, 'w') as f: f.write(version_response.get('data', '')) return True except Exception as e: print(f"Failed to backup {device['name']}: {e}") return False def compress_old_backups(backup_dir: Path, days: int): """Сжать резервные копии старше N дней""" cutoff_date = datetime.now() - timedelta(days=days) for backup_file in backup_dir.rglob("*.boot"): # Проверить возраст файла file_time = datetime.fromtimestamp(backup_file.stat().st_mtime) if file_time < cutoff_date and not backup_file.name.endswith('.gz'): # Сжать файл with open(backup_file, 'rb') as f_in: with gzip.open(f"{backup_file}.gz", 'wb') as f_out: shutil.copyfileobj(f_in, f_out) # Удалить оригинал backup_file.unlink() print(f"Compressed {backup_file}") def rotate_backups(backup_dir: Path, retention_days: int): """Удалить резервные копии старше retention_days""" cutoff_date = datetime.now() - timedelta(days=retention_days) for backup_file in backup_dir.rglob("config-*"): file_time = datetime.fromtimestamp(backup_file.stat().st_mtime) if file_time < cutoff_date: backup_file.unlink() print(f"Deleted old backup {backup_file}") def main(): """Главная функция""" print(f"=== VyOS Backup Script ===") print(f"Started: {datetime.now()}") # Создать директорию для резервных копий BACKUP_DIR.mkdir(parents=True, exist_ok=True) # Создать резервные копии success_count = 0 for device in DEVICES: if backup_device(device, BACKUP_DIR): success_count += 1 print(f"\nBackup completed: {success_count}/{len(DEVICES)} successful") # Сжать старые резервные копии print(f"\nCompressing backups older than {COMPRESSION_AFTER_DAYS} days...") compress_old_backups(BACKUP_DIR, COMPRESSION_AFTER_DAYS) # Ротация резервных копий print(f"\nRotating backups older than {RETENTION_DAYS} days...") rotate_backups(BACKUP_DIR, RETENTION_DAYS) print(f"\nFinished: {datetime.now()}") if __name__ == "__main__": main() ``` #### Cron задача для автоматического запуска ```bash # /etc/cron.d/vyos-backup # Ежедневное резервное копирование в 2:00 AM 0 2 * * * root /opt/scripts/backup_vyos.py >> /var/log/vyos-backup.log 2>&1 ``` ### Сценарий 4: CI/CD pipeline для автоматического тестирования конфигурации ```python #!/usr/bin/env python3 """ Автоматическое тестирование конфигурации VyOS """ import sys from vyos_client import VyOSClient from typing import List, Dict, Any class VyOSConfigTest: """Класс для тестирования конфигурации VyOS""" def __init__(self, host: str, api_key: str): self.client = VyOSClient(host=host, api_key=api_key, verify_ssl=False) self.test_results = [] def test_interface_status(self, interface: str) -> bool: """Проверить статус интерфейса""" try: result = self.client.show(path=["interfaces", interface]) if result.get('success'): self.test_results.append({ "test": f"Interface {interface} status", "status": "PASS", "details": "Interface is up" }) return True except: pass self.test_results.append({ "test": f"Interface {interface} status", "status": "FAIL", "details": "Interface is down or not found" }) return False def test_routing_table(self, expected_routes: List[str]) -> bool: """Проверить наличие маршрутов""" try: result = self.client.show(path=["ip", "route"]) if not result.get('success'): self.test_results.append({ "test": "Routing table", "status": "FAIL", "details": "Cannot retrieve routing table" }) return False routing_data = result.get('data', '') all_routes_present = True for route in expected_routes: if route not in routing_data: self.test_results.append({ "test": f"Route {route}", "status": "FAIL", "details": f"Route {route} not found" }) all_routes_present = False else: self.test_results.append({ "test": f"Route {route}", "status": "PASS", "details": "Route present" }) return all_routes_present except Exception as e: self.test_results.append({ "test": "Routing table", "status": "ERROR", "details": str(e) }) return False def test_service_running(self, service: str) -> bool: """Проверить работу сервиса""" try: result = self.client.show(path=["service", service]) if result.get('success'): self.test_results.append({ "test": f"Service {service}", "status": "PASS", "details": "Service is running" }) return True except: pass self.test_results.append({ "test": f"Service {service}", "status": "FAIL", "details": f"Service {service} is not running" }) return False def test_vpn_tunnels(self, expected_tunnels: int) -> bool: """Проверить количество активных VPN туннелей""" try: result = self.client.show(path=["vpn", "ipsec", "sa"]) if not result.get('success'): self.test_results.append({ "test": "VPN tunnels", "status": "FAIL", "details": "Cannot retrieve VPN status" }) return False # Парсинг вывода для подсчета туннелей (упрощенно) vpn_data = result.get('data', '') tunnel_count = vpn_data.count('ESTABLISHED') if tunnel_count >= expected_tunnels: self.test_results.append({ "test": "VPN tunnels", "status": "PASS", "details": f"{tunnel_count} tunnels established (expected >= {expected_tunnels})" }) return True else: self.test_results.append({ "test": "VPN tunnels", "status": "FAIL", "details": f"Only {tunnel_count} tunnels established (expected >= {expected_tunnels})" }) return False except Exception as e: self.test_results.append({ "test": "VPN tunnels", "status": "ERROR", "details": str(e) }) return False def test_connectivity(self, target: str) -> bool: """Проверить connectivity через ping""" try: result = self.client.show(path=["ping", target, "count", "3"]) if result.get('success'): ping_data = result.get('data', '') if "0% packet loss" in ping_data or "0.0% packet loss" in ping_data: self.test_results.append({ "test": f"Connectivity to {target}", "status": "PASS", "details": "Ping successful" }) return True except: pass self.test_results.append({ "test": f"Connectivity to {target}", "status": "FAIL", "details": f"Cannot reach {target}" }) return False def print_results(self): """Вывести результаты тестов""" print("\n=== Test Results ===") pass_count = sum(1 for r in self.test_results if r['status'] == 'PASS') fail_count = sum(1 for r in self.test_results if r['status'] == 'FAIL') error_count = sum(1 for r in self.test_results if r['status'] == 'ERROR') for result in self.test_results: status_icon = { 'PASS': '✓', 'FAIL': '✗', 'ERROR': '⚠' }.get(result['status'], '?') print(f"{status_icon} {result['test']}: {result['status']}") print(f" {result['details']}") print(f"\nSummary: {pass_count} passed, {fail_count} failed, {error_count} errors") print(f"Total tests: {len(self.test_results)}") # Return exit code return 0 if fail_count == 0 and error_count == 0 else 1 def main(): """Главная функция""" # Конфигурация тестов ROUTER_HOST = "192.168.1.1" API_KEY = "your-api-key-here" # Создать тестовый объект tester = VyOSConfigTest(ROUTER_HOST, API_KEY) # Запустить тесты print("Running VyOS configuration tests...") # Тест интерфейсов tester.test_interface_status("eth0") tester.test_interface_status("eth1") # Тест маршрутизации tester.test_routing_table([ "0.0.0.0/0", # Default route "192.168.1.0/24" # LAN route ]) # Тест сервисов tester.test_service_running("dhcp-server") tester.test_service_running("ssh") tester.test_service_running("ntp") # Тест VPN tester.test_vpn_tunnels(expected_tunnels=2) # Тест connectivity tester.test_connectivity("8.8.8.8") tester.test_connectivity("192.168.1.1") # Вывести результаты и вернуть exit code exit_code = tester.print_results() sys.exit(exit_code) if __name__ == "__main__": main() ``` ### Сценарий 5: Мониторинг и алертинг через API ```python #!/usr/bin/env python3 """ Мониторинг VyOS через API с отправкой алертов """ import smtplib import json from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from vyos_client import VyOSClient from dataclasses import dataclass from typing import List, Optional @dataclass class Alert: """Класс для представления алерта""" severity: str # critical, warning, info device: str message: str details: Optional[dict] = None class VyOSMonitor: """Класс для мониторинга VyOS устройств""" def __init__(self, devices: List[dict], api_key: str): self.devices = devices self.api_key = api_key self.alerts: List[Alert] = [] def check_interface_status(self, client: VyOSClient, device_name: str): """Проверить статус интерфейсов""" try: result = client.show(path=["interfaces"]) if not result.get('success'): self.alerts.append(Alert( severity="critical", device=device_name, message="Cannot retrieve interface status" )) return # Парсинг статуса интерфейсов (упрощенно) interfaces_data = result.get('data', '') # Проверить критичные интерфейсы critical_interfaces = ['eth0', 'eth1'] for iface in critical_interfaces: if f"{iface}:" in interfaces_data: if "down" in interfaces_data.lower(): self.alerts.append(Alert( severity="critical", device=device_name, message=f"Interface {iface} is DOWN" )) except Exception as e: self.alerts.append(Alert( severity="critical", device=device_name, message=f"Error checking interfaces: {e}" )) def check_vpn_tunnels(self, client: VyOSClient, device_name: str, expected_tunnels: int): """Проверить VPN туннели""" try: result = client.show(path=["vpn", "ipsec", "sa"]) if not result.get('success'): self.alerts.append(Alert( severity="warning", device=device_name, message="Cannot retrieve VPN status" )) return vpn_data = result.get('data', '') established_count = vpn_data.count('ESTABLISHED') if established_count < expected_tunnels: self.alerts.append(Alert( severity="warning", device=device_name, message=f"VPN tunnels degraded: {established_count}/{expected_tunnels}", details={"expected": expected_tunnels, "actual": established_count} )) except Exception as e: self.alerts.append(Alert( severity="warning", device=device_name, message=f"Error checking VPN: {e}" )) def check_system_resources(self, client: VyOSClient, device_name: str): """Проверить системные ресурсы""" try: # CPU cpu_result = client.show(path=["system", "cpu"]) if cpu_result.get('success'): cpu_data = cpu_result.get('data', '') # Упрощенный парсинг CPU usage # В реальности нужен более сложный парсинг if "CPU" in cpu_data: # Placeholder для демонстрации pass # Memory mem_result = client.show(path=["system", "memory"]) if mem_result.get('success'): mem_data = mem_result.get('data', '') # Упрощенный парсинг memory usage # В реальности нужен более сложный парсинг if "Mem:" in mem_data: # Placeholder для демонстрации pass except Exception as e: self.alerts.append(Alert( severity="info", device=device_name, message=f"Error checking system resources: {e}" )) def monitor_all(self): """Мониторить все устройства""" for device in self.devices: try: client = VyOSClient( host=device['host'], api_key=self.api_key, verify_ssl=False ) # Запустить проверки self.check_interface_status(client, device['name']) if device.get('expected_vpn_tunnels'): self.check_vpn_tunnels( client, device['name'], device['expected_vpn_tunnels'] ) self.check_system_resources(client, device['name']) except Exception as e: self.alerts.append(Alert( severity="critical", device=device['name'], message=f"Cannot connect to device: {e}" )) def send_alerts(self, smtp_config: dict): """Отправить алерты по email""" if not self.alerts: print("No alerts to send") return # Группировать алерты по severity critical = [a for a in self.alerts if a.severity == "critical"] warning = [a for a in self.alerts if a.severity == "warning"] info = [a for a in self.alerts if a.severity == "info"] # Формировать email msg = MIMEMultipart('alternative') msg['Subject'] = f"VyOS Monitoring Alert - {len(critical)} critical, {len(warning)} warnings" msg['From'] = smtp_config['from'] msg['To'] = smtp_config['to'] # Текст письма text = "VyOS Monitoring Alerts\n\n" if critical: text += "CRITICAL ALERTS:\n" for alert in critical: text += f" - {alert.device}: {alert.message}\n" text += "\n" if warning: text += "WARNING ALERTS:\n" for alert in warning: text += f" - {alert.device}: {alert.message}\n" text += "\n" if info: text += "INFO ALERTS:\n" for alert in info: text += f" - {alert.device}: {alert.message}\n" # HTML версия html = "" html += "

VyOS Monitoring Alerts

" if critical: html += "

CRITICAL ALERTS

" if warning: html += "

WARNING ALERTS

" if info: html += "

INFO ALERTS

" html += "" # Attach parts part1 = MIMEText(text, 'plain') part2 = MIMEText(html, 'html') msg.attach(part1) msg.attach(part2) # Отправить email try: with smtplib.SMTP(smtp_config['server'], smtp_config['port']) as server: if smtp_config.get('use_tls'): server.starttls() if smtp_config.get('username'): server.login(smtp_config['username'], smtp_config['password']) server.send_message(msg) print(f"Sent alert email with {len(self.alerts)} alerts") except Exception as e: print(f"Failed to send alert email: {e}") def main(): """Главная функция""" # Конфигурация устройств DEVICES = [ { "host": "192.168.1.1", "name": "border-router", "expected_vpn_tunnels": 3 }, { "host": "192.168.2.1", "name": "branch-01", "expected_vpn_tunnels": 1 }, ] # SMTP конфигурация SMTP_CONFIG = { "server": "smtp.example.com", "port": 587, "use_tls": True, "username": "monitoring@example.com", "password": "password", "from": "monitoring@example.com", "to": "admin@example.com" } API_KEY = "your-api-key-here" # Создать монитор monitor = VyOSMonitor(DEVICES, API_KEY) # Запустить мониторинг print("Starting VyOS monitoring...") monitor.monitor_all() # Отправить алерты monitor.send_alerts(SMTP_CONFIG) print("Monitoring completed") if __name__ == "__main__": main() ``` #### Cron задача для периодического мониторинга ```bash # /etc/cron.d/vyos-monitoring # Мониторинг каждые 5 минут */5 * * * * root /opt/scripts/monitor_vyos.py >> /var/log/vyos-monitor.log 2>&1 ``` ### Сценарий 6: Оркестрация множественных устройств с Ansible ```yaml --- # playbooks/orchestrate_network_change.yml - name: Orchestrate Network-Wide Configuration Change hosts: localhost gather_facts: false vars: change_description: "Add new VLAN 40 for IoT devices" vlan_id: 40 vlan_description: "IoT Devices" subnet: "192.168.40.0/24" gateway: "192.168.40.1" dhcp_start: "192.168.40.100" dhcp_stop: "192.168.40.200" tasks: - name: Log change start debug: msg: "Starting network change: {{ change_description }}" - name: Create pre-change backup include_tasks: backup_all_devices.yml - name: Apply VLAN configuration to core routers include_tasks: configure_vlan_core.yml vars: target_hosts: core_routers - name: Wait for core routers to stabilize pause: seconds: 30 - name: Verify core router configuration include_tasks: verify_vlan_config.yml vars: target_hosts: core_routers - name: Apply VLAN configuration to edge routers include_tasks: configure_vlan_edge.yml vars: target_hosts: edge_routers - name: Verify edge router configuration include_tasks: verify_vlan_config.yml vars: target_hosts: edge_routers - name: Run connectivity tests include_tasks: test_connectivity.yml - name: Log change completion debug: msg: "Network change completed successfully: {{ change_description }}" ``` ## Лучшие практики ### API Automation 1. **Используйте контекстные менеджеры** для автоматического commit/rollback 2. **Всегда включайте error handling** с логированием ошибок 3. **Используйте параллелизм** для операций на множестве устройств 4. **Кэшируйте клиенты** для повторного использования соединений 5. **Проверяйте success в ответах** перед обработкой данных 6. **Используйте SSL/TLS** в production окружениях 7. **Ротируйте API ключи** регулярно 8. **Логируйте все операции** для аудита 9. **Используйте batch операции** где возможно для оптимизации 10. **Тестируйте на staging** перед применением на production ### Ansible Best Practices 1. **Используйте Ansible Vault** для хранения секретов 2. **Группируйте хосты** в inventory по ролям и локациям 3. **Создавайте переиспользуемые роли** для общих задач 4. **Всегда используйте tags** для гибкого выполнения 5. **Включайте verify tasks** после конфигурационных изменений 6. **Используйте check mode** (`--check`) для dry-run 7. **Документируйте переменные** в README ролей 8. **Версионируйте playbooks** в Git 9. **Используйте динамический inventory** для больших парков 10. **Логируйте выполнение** playbooks для аудита ### Configuration Management 1. **Храните конфигурации в Git** с осмысленными commit messages 2. **Используйте branches** для разных окружений (dev/staging/prod) 3. **Автоматизируйте резервное копирование** с ротацией 4. **Тестируйте конфигурации** перед применением 5. **Документируйте изменения** в CHANGELOG 6. **Используйте CI/CD** для автоматического развертывания 7. **Проводите code review** для конфигурационных изменений 8. **Мониторьте состояние устройств** после изменений 9. **Имейте rollback plan** для критичных изменений 10. **Обучайте команду** работе с системой автоматизации ### Security 1. **Никогда не храните API ключи** в исходном коде 2. **Используйте переменные окружения** или секретные менеджеры 3. **Ограничивайте сетевой доступ** к API интерфейсам 4. **Используйте RBAC** для разграничения прав 5. **Аудируйте все API операции** в логах 6. **Ротируйте credentials** регулярно 7. **Используйте TLS** для всех API соединений 8. **Проверяйте SSL сертификаты** в production 9. **Ограничивайте rate limiting** для API запросов 10. **Мониторьте подозрительную активность** через API ## Troubleshooting ### Проблема: API запросы возвращают 401 Unauthorized **Причина**: Неверный API ключ или API не настроен **Решение**: ```bash # Проверить конфигурацию API show configuration service https api # Убедиться что API включен show service https # Для VyOS 1.5 проверить REST endpoint show configuration service https api rest # Создать новый API ключ configure set service https api keys id mykey key 'new-api-key' commit save ``` ### Проблема: Ansible playbook fails with "Network device unreachable" **Причина**: Неверная конфигурация connection или SSH **Решение**: ```yaml # Проверить inventory конфигурацию ansible_connection: ansible.netcommon.network_cli ansible_network_os: vyos.vyos.vyos # Тестировать connectivity ansible -m ping -i inventory/hosts.yml vyos_routers # Debug Ansible connection ansible-playbook playbook.yml -vvv ``` ### Проблема: Configuration не применяется через API **Причина**: Не выполнен commit или есть ошибки в конфигурации **Решение**: ```python # Всегда проверять success в ответах response = client.set_config(path=["interfaces", "ethernet", "eth1", "address"], value="192.168.1.1/24") if not response.get('success'): print(f"Error: {response.get('error')}") # Обязательно commit после изменений client.commit() client.save() # Проверить конфигурацию config = client.show_config(path=["interfaces", "ethernet", "eth1"]) print(config) ``` ### Проблема: Batch операции выполняются частично **Причина**: Ошибка в одной из команд прерывает всю транзакцию **Решение**: ```python # Использовать try/except для каждой операции commands = [ {"op": "set", "path": ["interfaces", "ethernet", "eth1", "address"], "value": "192.168.1.1/24"}, {"op": "set", "path": ["interfaces", "ethernet", "eth2", "address"], "value": "192.168.2.1/24"}, ] for cmd in commands: try: result = client.set_config(cmd['path'], cmd.get('value')) if not result.get('success'): print(f"Warning: Failed to apply {cmd['path']}: {result.get('error')}") except Exception as e: print(f"Error: {e}") # Commit только если все успешно client.commit() ``` ### Проблема: SSL Certificate Verification Failed **Причина**: Self-signed сертификат на VyOS **Решение**: ```python # Отключить проверку SSL для testing/development client = VyOSClient( host="192.168.1.1", api_key="your-key", verify_ssl=False ) # В production - установить правильный сертификат на VyOS # или добавить CA в trust store import requests requests.packages.urllib3.disable_warnings() ``` ## Заключение Автоматизация VyOS через REST API, Ansible и Python предоставляет мощные инструменты для: - **Масштабирования**: Управление парком устройств из единого центра - **Консистентности**: Единообразная конфигурация всех устройств - **Скорости**: Быстрое развертывание изменений - **Надежности**: Автоматизированное тестирование и rollback - **Аудита**: Полное логирование всех операций - **DevOps интеграции**: CI/CD для сетевой инфраструктуры Используйте эти инструменты для построения современной, гибкой и надежной сетевой инфраструктуры на базе VyOS.