I have the same setup, I'm currently deploying 144 SG1535 with dual wan with public fixed ips. Usually DAIP solves this issue, but with DAIP I lose the capability of identity sharing, because of this I changed from DAIP to a standard gateway.
I made a custom logic with python and systemd in my sms that does the following:
1 - Using mgmt api, the script fetches the gateway description, the description contains a specific format with the public ips of both wan links of the gateway.
Description format: ID:SITEIDNAME;WANMON:ENABLED;WAN1IP:X.X.X.1;WAN2IP:X.X.X.2
2 - The script parses the description and checks if WANMON is enabled; it then uses the wan ips for the next steps.
3 - The script check if sic is initialized, if not bypasses the gateway.4 - The script is periodically testing both wan links with a tcp probe on port 18191
4 - It always considers the wan1 as primary, and if wan1 fails and wan2 is up, it uses the mgmt api to change the ip of the gateway from wan1 to wan2, after wan1 is up it changes from wan2 to wan1.
The script saves a log in the filesystem and rotates the log file when it reaches 50MB.
Systemd service
[Unit]
Description=Check Point Gateway WAN Monitor
After=network.target
[Service]
Type=simple
ExecStart=/bin/bash -l -c "exec /var/log/gw_monitor/gw_monitor.py"
Restart=on-failure
RestartSec=10
[Install]
WantedBy=multi-user.target
Python script
#!/usr/bin/env python3
import os
import json
import time
import socket
import re
import concurrent.futures
import subprocess
import gzip
import shutil
import datetime
# --- Variables ---
VERSION = "1.1.0"
EXEC_DIR = os.path.dirname(os.path.abspath(__file__))
OUTPUT_DIR = os.path.join(EXEC_DIR, "data", "output")
LOG_FILE = os.path.join(OUTPUT_DIR, "gw_monitor.log")
SESSION_FILE = os.path.join(OUTPUT_DIR, "session.txt")
MAX_LOG_SIZE = 50 * 1024 * 1024 # 50MB
os.makedirs(OUTPUT_DIR, exist_ok=True)
# Regex to parse the comments field
REGEX_WANMON = re.compile(r"WANMON:\s*ENABLED", re.IGNORECASE)
REGEX_WAN1 = re.compile(r"WAN1IP:\s*([0-9\.]+)", re.IGNORECASE)
REGEX_WAN2 = re.compile(r"WAN2IP:\s*([0-9\.]+)", re.IGNORECASE)
def log_msg(msg):
"""Log messages to both console and a log file with rotation and compression."""
timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
formatted = f"[{timestamp}] {msg}"
print(formatted)
# Check if rotation is needed
try:
if os.path.exists(LOG_FILE) and os.path.getsize(LOG_FILE) > MAX_LOG_SIZE:
rotate_logs()
except Exception as e:
print(f"Error checking log size: {e}")
with open(LOG_FILE, "a") as f:
f.write(formatted + "\n")
def rotate_logs():
"""Compress the current log file and rename it using the timestamp of the oldest log found within."""
if not os.path.exists(LOG_FILE):
return
try:
# Default timestamp (current) in case file is empty or unparseable
oldest_ts = datetime.datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
# Read the first line to find the oldest timestamp
with open(LOG_FILE, 'r') as f:
first_line = f.readline()
match = re.search(r'\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\]', first_line)
if match:
# Format to be filename friendly
oldest_ts = match.group(1).replace(" ", "_").replace(":", "-")
# Construction of the archive filename
archive_name = LOG_FILE.replace(".log", "") + f"_{oldest_ts}.log.gz"
# Compress the file
with open(LOG_FILE, 'rb') as f_in:
with gzip.open(archive_name, 'wb') as f_out:
shutil.copyfileobj(f_in, f_out)
# Clear the original log file
with open(LOG_FILE, 'w') as f:
f.truncate()
print(f"[Log Rotated: {archive_name}]")
except Exception as e:
print(f"CRITICAL: Failed to rotate logs: {e}")
def check_tcp_port(gw_dict, port=18191, timeout=3, retries=3, delay=2):
"""
Checks the active IP for a single gateway with robust retry logic.
Returns the gateway dictionary and the reachable IP (or None if down).
"""
wan1 = gw_dict.get("wan1_ip")
wan2 = gw_dict.get("wan2_ip")
def is_reachable(ip):
if not ip:
return False
for attempt in range(retries):
try:
with socket.create_connection((ip, port), timeout=timeout):
return True
except (socket.timeout, socket.error):
if attempt < retries - 1:
time.sleep(delay)
return False
if is_reachable(wan1):
return gw_dict, wan1
if is_reachable(wan2):
return gw_dict, wan2
return gw_dict, None
def run_mgmt_cli(command_list):
"""Run a mgmt_cli command using subprocess and return status code and parsed JSON."""
try:
full_cmd = ["mgmt_cli"] + command_list + ["-s", SESSION_FILE, "-f", "json"]
result = subprocess.run(full_cmd, capture_output=True, text=True)
if result.returncode == 0:
return 200, json.loads(result.stdout)
else:
try:
# mgmt_cli often outputs valid JSON errors when -f json is used
return result.returncode, json.loads(result.stdout)
except json.JSONDecodeError:
return result.returncode, {"error": result.stderr or result.stdout}
except Exception as e:
return -1, {"error": str(e)}
def run_sic_check(gw):
"""Check SIC status for a gateway object."""
status, res = run_mgmt_cli(["test-sic-status", "name", gw["name"]])
if status == 200:
return gw, res.get("sic-status", "unknown")
return gw, "error"
def discover_gateways():
"""Query SMS to find gateways with WANMON:ENABLED and verified SIC."""
all_candidates = []
offset = 0
limit = 500
while True:
cmd = ["show", "gateways-and-servers", "limit", str(limit), "offset", str(offset), "details-level", "full"]
status, res = run_mgmt_cli(cmd)
if status != 200:
log_msg(f"ERROR: Failed to discover gateways: {json.dumps(res)}")
break
objects = res.get("objects", [])
if not objects:
break
for obj in objects:
if obj.get("type") not in ("simple-gateway", "CpmiGatewayPlain"):
continue
comments = obj.get("comments", "")
if comments and REGEX_WANMON.search(comments):
w1_match = REGEX_WAN1.search(comments)
w2_match = REGEX_WAN2.search(comments)
gw = {
"uid": obj.get("uid"),
"name": obj.get("name"),
"current_ip": obj.get("ipv4-address"),
"wan1_ip": w1_match.group(1) if w1_match else None,
"wan2_ip": w2_match.group(1) if w2_match else None
}
if gw["wan1_ip"]:
all_candidates.append(gw)
else:
log_msg(f"WARNING: Gateway '{gw['name']}' has WANMON:ENABLED but no WAN1IP found in comments.")
total = res.get("total", 0)
offset += limit
if offset >= total:
break
if not all_candidates:
return []
log_msg(f"Performing SIC verification for {len(all_candidates)} candidates...")
monitored_gateways = []
# Use parallel threads to speed up SIC checks
with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor:
future_to_gw = {executor.submit(run_sic_check, c): c for c in all_candidates}
for future in concurrent.futures.as_completed(future_to_gw):
gw, sic_status = future.result()
# We allow 'communicating' and 'not communicating'.
# 'not communicating' often means the link is down, which is exactly when we want to failover.
# But 'uninitialized' means the gateway isn't ready at all.
if sic_status in ["communicating", "not communicating", "not-communicating"]:
monitored_gateways.append(gw)
else:
log_msg(f"INFO: Gateway '{gw['name']}' skipped (SIC: {sic_status}).")
return monitored_gateways
def main():
log_msg("--- Gateways Monitor Started ---")
login_cmd = [
"mgmt_cli", "login", "-r", "true",
"session-name", "WAN Monitor Failover System",
"session-description", "Automated IP route updates"
]
login_result = subprocess.run(login_cmd, capture_output=True, text=True)
if login_result.returncode != 0:
log_msg(f"ERROR: mgmt_cli login failed: {login_result.stderr or login_result.stdout}")
return
# Write the standard text stdout directly to the session file
# `mgmt_cli` expects this specific multi-line format to parse the session correctly!
with open(SESSION_FILE, "w") as f:
f.write(login_result.stdout)
log_msg("Login successful. Root session established.")
try:
# 2. Dynamic Discovery
log_msg("Discovering gateways with WANMON:ENABLED...")
gateways = discover_gateways()
if not gateways:
log_msg("No gateways found with WANMON:ENABLED in comments. Exiting.")
return
log_msg(f"Found {len(gateways)} gateways to monitor.")
# 3. Concurrent TCP Reachability Check
log_msg("Starting concurrent reachability checks...")
to_update = []
with concurrent.futures.ThreadPoolExecutor(max_workers=150) as executor:
future_to_gw = {executor.submit(check_tcp_port, gw): gw for gw in gateways}
for future in concurrent.futures.as_completed(future_to_gw):
gw, active_ip = future.result()
name = gw["name"]
if not active_ip:
log_msg(f"[{name}] UNREACHABLE on both WAN1IP and WAN2IP.")
continue
if active_ip == gw["current_ip"]:
pass # Already matching
else:
log_msg(f"[{name}] IP change detected: {gw['current_ip']} -> {active_ip}")
gw["new_ip"] = active_ip
to_update.append(gw)
# 4. Check and Update Gateways via mgmt_cli
if to_update:
log_msg(f"Applying {len(to_update)} updates to Management Server...")
changes_made = False
for gw in to_update:
set_cmd = [
"set", "simple-gateway",
"name", gw["name"],
"ipv4-address", gw["new_ip"],
"ignore-warnings", "true"
]
set_status, set_res = run_mgmt_cli(set_cmd)
if set_status == 200:
log_msg(f"Successfully updated '{gw['name']}'.")
changes_made = True
else:
log_msg(f"ERROR: Failed to update '{gw['name']}': {json.dumps(set_res)}")
# 5. Publish Changes
if changes_made:
log_msg("Publishing changes to SMS database...")
pub_status, pub_res = run_mgmt_cli(["publish"])
if pub_status == 200:
task_id = pub_res.get("task-id")
log_msg(f"Publish task started (Task ID: {task_id}). Changes will be active shortly.")
else:
log_msg(f"ERROR: Publish failed: {json.dumps(pub_res)}")
else:
log_msg("No updates required. All gateways on active IP.")
finally:
# 6. Logout
log_msg("Logging out from Check Point API...")
subprocess.run(["mgmt_cli", "logout", "-s", SESSION_FILE, "-f", "json"], capture_output=True)
# Clean up the session file
if os.path.exists(SESSION_FILE):
os.remove(SESSION_FILE)
log_msg("--- Gateways Monitor Finished ---\n")
if __name__ == "__main__":
log_msg(f"Starting Gateway Monitor v{VERSION} - Service Mode (systemd).")
while True:
try:
main()
except Exception as e:
log_msg(f"FATAL ERROR in main loop: {e}")
time.sleep(60)
