VyOS Automation - REST API, Ansible and Python

configure

# Create an API key
set service https api keys id automation key 'your-api-key-here'

# Enable the REST API
set service https api rest

# Optional: configure the HTTPS certificate
set service https certificates ca-certificate ca-vyos
set service https certificates certificate vyos-cert

# Configure access
set service https listen-address 192.168.1.1

commit
save

configure

# Version 1.4 uses a single API endpoint
set service https api keys id automation key 'your-api-key-here'

commit
save

POST /configure
Content-Type: application/json

{
  "op": "set",
  "path": ["interfaces", "ethernet", "eth1", "address"],
  "value": "192.168.2.1/24",
  "key": "your-api-key-here"
}

POST /configure
Content-Type: application/json

{
  "op": "delete",
  "path": ["interfaces", "ethernet", "eth1", "address", "192.168.2.1/24"],
  "key": "your-api-key-here"
}

POST /retrieve
Content-Type: application/json

{
  "op": "showConfig",
  "path": ["interfaces", "ethernet"],
  "key": "your-api-key-here"
}

POST /show
Content-Type: application/json

{
  "op": "show",
  "path": ["interfaces"],
  "key": "your-api-key-here"
}

POST /config-file
Content-Type: application/json

{
  "op": "save",
  "key": "your-api-key-here"
}

POST /config-file
Content-Type: application/json

{
  "op": "load",
  "file": "/config/config.boot.backup",
  "key": "your-api-key-here"
}

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"
}

POST /image
Content-Type: application/json

{
  "op": "delete",
  "name": "1.4.20230101",
  "key": "your-api-key-here"
}

POST /reboot
Content-Type: application/json

{
  "key": "your-api-key-here"
}

POST /poweroff
Content-Type: application/json

{
  "key": "your-api-key-here"
}

import requests
import json
from typing import Dict, List, Any, Optional

class VyOSClient:
    """Client for working with the 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]:
        """Perform an API request"""
        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]:
        """Set a configuration value"""
        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]:
        """Delete a configuration element"""
        data = {
            "op": "delete",
            "path": path
        }
        return self._make_request("configure", data)

    def show_config(self, path: List[str] = []) -> Dict[str, Any]:
        """Retrieve the configuration"""
        data = {
            "op": "showConfig",
            "path": path
        }
        return self._make_request("retrieve", data)

    def show(self, path: List[str]) -> Dict[str, Any]:
        """Run a show command"""
        data = {
            "op": "show",
            "path": path
        }
        return self._make_request("show", data)

    def commit(self) -> Dict[str, Any]:
        """Apply the configuration"""
        data = {"op": "commit"}
        return self._make_request("configure", data)

    def save(self) -> Dict[str, Any]:
        """Save the configuration"""
        data = {"op": "save"}
        return self._make_request("config-file", data)

    def batch_configure(self, commands: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Perform a batch configuration"""
        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)

        # Apply the changes
        commit_result = self.commit()
        results.append(commit_result)

        # Save
        save_result = self.save()
        results.append(save_result)

        return {"results": results}

# Usage example
if __name__ == "__main__":
    # Create the client
    client = VyOSClient(
        host="192.168.1.1",
        api_key="your-api-key-here",
        verify_ssl=False  # For self-signed certificates
    )

    # Configure the interface
    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"
    )

    # Apply and save
    client.commit()
    client.save()

    # Retrieve the interface configuration
    config = client.show_config(path=["interfaces", "ethernet", "eth1"])
    print(json.dumps(config, indent=2))

    # Retrieve interface status
    status = client.show(path=["interfaces"])
    print(json.dumps(status, indent=2))

from contextlib import contextmanager
import logging

class VyOSConfigSession(VyOSClient):
    """Client with automatic commit/save and rollback on errors"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.logger = logging.getLogger(__name__)

    @contextmanager
    def config_session(self, auto_save: bool = True):
        """Context manager for safe configuration"""
        try:
            yield self
            # On success - commit and save
            self.commit()
            if auto_save:
                self.save()
            self.logger.info("Configuration applied successfully")
        except Exception as e:
            # On error - 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

# Usage example
with VyOSConfigSession(
    host="192.168.1.1",
    api_key="your-api-key-here",
    verify_ssl=False
).config_session() as session:
    # All changes in this block are automatically committed/saved
    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"
    )
    # On exiting the block - automatic commit and save
    # On error - automatic rollback

# Install Ansible
pip install ansible

# Install the VyOS collection
ansible-galaxy collection install vyos.vyos

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

# Create an encrypted file with passwords
ansible-vault create inventory/group_vars/vyos_routers/vault.yml

# Contents of vault.yml
vault_vyos_password: "secure-password-here"

---
# 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

---
# 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

---
# 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/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

---
# 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"

---
# 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

---
- 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

---
# 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

#!/usr/bin/env python3
# scripts/deploy.py

import sys
import argparse
from pathlib import Path
import subprocess
from vyos_client import VyOSClient  # From the example above

def deploy_config(device: str, config_file: Path, api_key: str):
    """Deploy configuration to a device"""

    # Read the configuration
    with open(config_file) as f:
        config_lines = f.readlines()

    # Create the client
    client = VyOSClient(host=device, api_key=api_key, verify_ssl=False)

    # Parse and apply the configuration
    for line in config_lines:
        line = line.strip()
        if not line or line.startswith('#'):
            continue

        if line.startswith('set '):
            # Extract path and value from the set command
            parts = line.split()[1:]  # Strip 'set'

            # Find the value (everything after the last space, if quotes are present)
            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 and 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()

# .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

#!/usr/bin/env python3
"""
Bulk interface configuration across multiple devices
"""

from vyos_client import VyOSClient
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# List of devices
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 configuration
INTERFACE_CONFIG = {
    "eth0": {
        "description": "WAN",
        "address": "dhcp"
    },
    "eth1": {
        "description": "LAN",
        "address": None,  # Set individually
        "mtu": "1500"
    }
}

API_KEY = "your-api-key-here"

def configure_device(device: dict) -> dict:
    """Configure a single device"""
    try:
        client = VyOSClient(
            host=device['host'],
            api_key=API_KEY,
            verify_ssl=False
        )

        logger.info(f"Configuring {device['name']} ({device['host']})")

        # Configure the interfaces
        for iface, config in INTERFACE_CONFIG.items():
            # Description
            if 'description' in config:
                client.set_config(
                    path=["interfaces", "ethernet", iface, "description"],
                    value=config['description']
                )

            # Address (if not 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 and 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():
    """Main function"""
    results = []

    # Configure devices in parallel
    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 the results
    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()

---
# 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]

#!/usr/bin/env python3
"""
Automated backup of VyOS configuration with rotation
"""

import os
import sys
from datetime import datetime, timedelta
from pathlib import Path
import gzip
import shutil
from vyos_client import VyOSClient

# Configuration
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:
    """Create a backup of a device configuration"""
    try:
        client = VyOSClient(
            host=device['host'],
            api_key=API_KEY,
            verify_ssl=False
        )

        # Retrieve the configuration
        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', '')

        # Create a directory for the device
        device_backup_dir = backup_dir / device['name']
        device_backup_dir.mkdir(parents=True, exist_ok=True)

        # File name with a timestamp
        timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
        backup_file = device_backup_dir / f"config-{timestamp}.boot"

        # Save the configuration
        with open(backup_file, 'w') as f:
            f.write(config_data)

        print(f"Backed up {device['name']} to {backup_file}")

        # Retrieve additional information
        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):
    """Compress backups older than N days"""
    cutoff_date = datetime.now() - timedelta(days=days)

    for backup_file in backup_dir.rglob("*.boot"):
        # Check the file age
        file_time = datetime.fromtimestamp(backup_file.stat().st_mtime)

        if file_time < cutoff_date and not backup_file.name.endswith('.gz'):
            # Compress the file
            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)

            # Delete the original
            backup_file.unlink()
            print(f"Compressed {backup_file}")

def rotate_backups(backup_dir: Path, retention_days: int):
    """Delete backups older than 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():
    """Main function"""
    print(f"=== VyOS Backup Script ===")
    print(f"Started: {datetime.now()}")

    # Create the backup directory
    BACKUP_DIR.mkdir(parents=True, exist_ok=True)

    # Create the backups
    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")

    # Compress old backups
    print(f"\nCompressing backups older than {COMPRESSION_AFTER_DAYS} days...")
    compress_old_backups(BACKUP_DIR, COMPRESSION_AFTER_DAYS)

    # Rotate the backups
    print(f"\nRotating backups older than {RETENTION_DAYS} days...")
    rotate_backups(BACKUP_DIR, RETENTION_DAYS)

    print(f"\nFinished: {datetime.now()}")

if __name__ == "__main__":
    main()

# /etc/cron.d/vyos-backup
# Daily backup at 2:00 AM
0 2 * * * root /opt/scripts/backup_vyos.py >> /var/log/vyos-backup.log 2>&1

#!/usr/bin/env python3
"""
Automated testing of VyOS configuration
"""

import sys
from vyos_client import VyOSClient
from typing import List, Dict, Any

class VyOSConfigTest:
    """Class for testing VyOS configuration"""

    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:
        """Check the interface status"""
        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:
        """Check for the presence of routes"""
        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:
        """Check that a service is running"""
        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:
        """Check the number of active VPN tunnels"""
        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

            # Parse the output to count tunnels (simplified)
            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:
        """Check connectivity via 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 the test results"""
        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():
    """Main function"""
    # Test configuration
    ROUTER_HOST = "192.168.1.1"
    API_KEY = "your-api-key-here"

    # Create the test object
    tester = VyOSConfigTest(ROUTER_HOST, API_KEY)

    # Run the tests
    print("Running VyOS configuration tests...")

    # Interface tests
    tester.test_interface_status("eth0")
    tester.test_interface_status("eth1")

    # Routing test
    tester.test_routing_table([
        "0.0.0.0/0",  # Default route
        "192.168.1.0/24"  # LAN route
    ])

    # Service tests
    tester.test_service_running("dhcp-server")
    tester.test_service_running("ssh")
    tester.test_service_running("ntp")

    # VPN test
    tester.test_vpn_tunnels(expected_tunnels=2)

    # Connectivity test
    tester.test_connectivity("8.8.8.8")
    tester.test_connectivity("192.168.1.1")

    # Print the results and return the exit code
    exit_code = tester.print_results()
    sys.exit(exit_code)

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Monitoring VyOS through the API with alert delivery
"""

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:
    """Class representing an alert"""
    severity: str  # critical, warning, info
    device: str
    message: str
    details: Optional[dict] = None

class VyOSMonitor:
    """Class for monitoring VyOS devices"""

    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):
        """Check the interface status"""
        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

            # Parse the interface status (simplified)
            interfaces_data = result.get('data', '')

            # Check critical interfaces
            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):
        """Check the VPN tunnels"""
        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):
        """Check system resources"""
        try:
            # CPU
            cpu_result = client.show(path=["system", "cpu"])
            if cpu_result.get('success'):
                cpu_data = cpu_result.get('data', '')
                # Simplified CPU usage parsing
                # In practice a more complex parser is required
                if "CPU" in cpu_data:
                    # Placeholder for demonstration
                    pass

            # Memory
            mem_result = client.show(path=["system", "memory"])
            if mem_result.get('success'):
                mem_data = mem_result.get('data', '')
                # Simplified memory usage parsing
                # In practice a more complex parser is required
                if "Mem:" in mem_data:
                    # Placeholder for demonstration
                    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):
        """Monitor all devices"""
        for device in self.devices:
            try:
                client = VyOSClient(
                    host=device['host'],
                    api_key=self.api_key,
                    verify_ssl=False
                )

                # Run the checks
                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):
        """Send alerts by email"""
        if not self.alerts:
            print("No alerts to send")
            return

        # Group alerts by 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"]

        # Build the 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']

        # Email text
        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 version
        html = "<html><body>"
        html += "<h2>VyOS Monitoring Alerts</h2>"

        if critical:
            html += "<h3 style='color: red;'>CRITICAL ALERTS</h3><ul>"
            for alert in critical:
                html += f"<li><strong>{alert.device}</strong>: {alert.message}</li>"
            html += "</ul>"

        if warning:
            html += "<h3 style='color: orange;'>WARNING ALERTS</h3><ul>"
            for alert in warning:
                html += f"<li><strong>{alert.device}</strong>: {alert.message}</li>"
            html += "</ul>"

        if info:
            html += "<h3>INFO ALERTS</h3><ul>"
            for alert in info:
                html += f"<li><strong>{alert.device}</strong>: {alert.message}</li>"
            html += "</ul>"

        html += "</body></html>"

        # Attach parts
        part1 = MIMEText(text, 'plain')
        part2 = MIMEText(html, 'html')
        msg.attach(part1)
        msg.attach(part2)

        # Send the 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():
    """Main function"""
    # Device configuration
    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 configuration
    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"

    # Create the monitor
    monitor = VyOSMonitor(DEVICES, API_KEY)

    # Run the monitoring
    print("Starting VyOS monitoring...")
    monitor.monitor_all()

    # Send the alerts
    monitor.send_alerts(SMTP_CONFIG)

    print("Monitoring completed")

if __name__ == "__main__":
    main()

# /etc/cron.d/vyos-monitoring
# Monitoring every 5 minutes
*/5 * * * * root /opt/scripts/monitor_vyos.py >> /var/log/vyos-monitor.log 2>&1

---
# 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 }}"

# Check the API configuration
show configuration service https api

# Make sure the API is enabled
show service https

# For VyOS 1.5, check the REST endpoint
show configuration service https api rest

# Create a new API key
configure
set service https api keys id mykey key 'new-api-key'
commit
save

# Check the inventory configuration
ansible_connection: ansible.netcommon.network_cli
ansible_network_os: vyos.vyos.vyos

# Test connectivity
ansible -m ping -i inventory/hosts.yml vyos_routers

# Debug the Ansible connection
ansible-playbook playbook.yml -vvv

# Always check success in the responses
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')}")

# Always commit after changes
client.commit()
client.save()

# Check the configuration
config = client.show_config(path=["interfaces", "ethernet", "eth1"])
print(config)

# Use try/except for each operation
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 only if everything succeeded
client.commit()

# Disable SSL verification for testing/development
client = VyOSClient(
    host="192.168.1.1",
    api_key="your-key",
    verify_ssl=False
)

# In production - install a proper certificate on VyOS
# or add the CA to the trust store
import requests
requests.packages.urllib3.disable_warnings()