I made this for this purpose. Place both of these in the /config/scripts
folder:
wgcheck.sh
#!/bin/vbash
source /opt/vyatta/etc/functions/script-template
source <(/config/scripts/wgcheck.py)
wgcheck.py
#!/usr/bin/env python3
import subprocess
import socket
import syslog
# list of tunnels. Format: [interface_name, peer_name, fqdn, ip_to_check, ip_version]
tunnels = [
["wg0", "TEST1", "l0crian1.test", "10.5.1.1", "ipv4"],
["wg1", "TEST2", "l0crian2.test", "10.5.1.5", "ipv4"]
]
def log_message(message, level=syslog.LOG_INFO):
# Open syslog
syslog.openlog("wgcheck.py", syslog.LOG_PID)
# Log the message
syslog.syslog(level, message)
# Close syslog (not necessary unless you need to change the identifier or options)
syslog.closelog()
def ping(ip, count=1, timeout=1):
"""Ping an IP address to check connectivity."""
try:
response = subprocess.run(
["ping", "-c", str(count), "-W", str(timeout), ip],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
return response.returncode == 0 # Return True if ping succeeds, else False
except Exception as e:
return False
def resolve_fqdn(fqdn, ip_version='ipv4'):
"""Resolve FQDN to IP using DNS based on IP version."""
family = socket.AF_INET if ip_version == 'ipv4' else socket.AF_INET6
try:
return socket.getaddrinfo(fqdn, None, family)[0][4][0]
except socket.gaierror as e:
return None
def check_tunnels(tunnels):
global new_ip
"""Check each tunnel and perform actions based on ping results."""
for i in tunnels:
if ping(i[3]):
pass
else:
# Get current configured endpoint address
try:
current_ip = subprocess.run(f"cli-shell-api showCfg --show-active-only | vyos-config-to-commands | grep 'set interfaces wireguard {i[0]} peer {i[1]} address'", cwd="/usr/libexec/vyos/op_mode/", stdout=subprocess.PIPE, shell=True, text=True).stdout.encode('utf-8').decode('utf-8').split()[-1].replace("'","")
except:
log_message(f"Could not find configured endpoint for {i[0]} peer {i[1]}", syslog.LOG_INFO)
continue
new_ip = resolve_fqdn(i[2], i[4])
if new_ip:
if new_ip != current_ip:
log_message(f"Updated endpoint to {new_ip} for {i[0]} peer {i[1]}", syslog.LOG_INFO)
config_list.append(i)
config_list = []
check_tunnels(tunnels)
if config_list:
print('configure')
for interface, peer_name, fqdn, ip_to_check, ip_version in config_list:
print(f"set interfaces wireguard {interface} peer {peer_name} address {new_ip}")
print('commit')
Here’s the task scheduler config:
Task Scheduler config
set system task-scheduler task wgcheck executable path '/config/scripts/wgcheck.sh'
set system task-scheduler task wgcheck interval '1m'
Make sure you make the scripts executable.
You need to populate the list on the top of the wgcheck.py script with any tunnel/peer you want to check. Just use the current list as an example.
This will check every minute. The script will perform these functions:
- Attempt to ping other side of tunnel
- If ping is successful, no further action is taken
- Check DNS of supplied FQDN
- If returned IP is different than configured IP, the configuration is updated