

ICARAX Tech Blog – How to programmatically verify and apply the critical Fireware OS code‑injection patch released by WatchGuard
TL;DR – This guide shows you how to use WatchGuard’s REST‑based Management API (available on Fireware OS v12.5+ and later) to:
- Pull the current firmware version of a managed device.
- Compare it against the known‑good patched version.
- Trigger a firmware upgrade (or schedule it) if the device is vulnerable.
The examples are ready‑to‑copy, include proper error handling, logging, and configuration via environment variables.
| Item | Why you need it | Version / Notes |
|---|---|---|
| WatchGuard Firebox / XTM | Device that runs Fireware OS and exposes the Management API | Fireware OS v12.5 or newer (API available from v12.5) |
| API credentials | Username/password or token for the Management API | Create a local admin or API‑only user in System → Administrators |
| Python 3.9+ (or Node.js 18+) | Runtime for the code samples | Official distributions |
| Package manager | pip for Python, npm or yarn for JS/TS | Comes with the runtime |
| HTTPS access | The Management API is TLS‑protected; you must trust the device’s cert or disable verification (not recommended for prod) | Use a CA‑signed cert or add the device cert to your trust store |
| (Optional) Git | To clone the example repo if you prefer | Any recent version |
Note: The WatchGuard Management API is read‑write – calling the upgrade endpoint will initiate a firmware flash. Test on a lab device first!
# Clone the repo (optional)
git clone https://github.com/icarax/watchguard-fireware-patch-demo.git
cd watchguard-fireware-patch-demo
# Create a virtual environment
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
# Install dependencies
pip install --upgrade pip
pip install httpx[http2] tqdm python-dotenv loguru
git clone https://github.com/icarax/watchguard-fireware-patch-demo.git
cd watchguard-fireware-patch-demo/js
# Initialize npm project (if not already)
npm init -y
# Install dependencies
npm install axios dotenv p-limit
# For TypeScript dev dependencies
npm install --save-dev typescript @types/node ts-node nodemon
Both language samples follow the same logical flow:
/api/v1/devices/{id}/firmware → current version.12.5.4 Build 100123)./api/v1/devices/{id}/firmware/upgrade with the upgrade payload.patch_fireware.py)#!/usr/bin/env python3
"""
WatchGuard Fireware OS patch automation – Python version.
Requires:
httpx, python-dotenv, loguru, tqdm
"""
import os
import sys
import time
from typing import Any, Dict
import httpx
from dotenv import load_dotenv
from loguru import logger
from tqdm import tqdm
# ----------------------------------------------------------------------
# Configuration (loaded from .env)
# ----------------------------------------------------------------------
load_dotenv() # pulls WG_API_URL, WG_USERNAME, WG_PASSWORD, WG_DEVICE_ID, WG_PATCHED_VERSION
API_BASE = os.getenv("WG_API_URL", "").rstrip("/")
USERNAME = os.getenv("WG_USERNAME")
PASSWORD = os.getenv("WG_PASSWORD")
DEVICE_ID = os.getenv("WG_DEVICE_ID") # numeric or UUID as shown in UI
PATCHED_VERSION = os.getenv("WG_PATCHED_VERSION") # e.g. "12.5.4 Build 100123"
TIMEOUT = int(os.getenv("WG_TIMEOUT", "30"))
POLL_INTERVAL = int(os.getenv("WG_POLL_INTERVAL", "15"))
MAX_RETRIES = int(os.getenv("WG_MAX_RETRIES", "12")) # ~3 minutes with 15s interval
if not all([API_BASE, USERNAME, PASSWORD, DEVICE_ID, PATCHED_VERSION]):
logger.error("Missing required environment variables. See .env.example")
sys.exit(1)
# ----------------------------------------------------------------------
# Helper: simple JWT‑like token retrieval (WatchGuard API)
# ----------------------------------------------------------------------
def get_auth_token(client: httpx.Client) -> str:
"""Exchange username/password for a Bearer token."""
auth_url = f"{API_BASE}/api/v1/auth/token"
resp = client.post(
auth_url,
json={"username": USERNAME, "password": PASSWORD},
timeout=TIMEOUT,
)
resp.raise_for_status()
data = resp.json()
token = data.get("access_token")
if not token:
raise RuntimeError("Auth response missing access_token")
return token
# ----------------------------------------------------------------------
# Core logic
# ----------------------------------------------------------------------
def fetch_current_version(client: httpx.Client, token: str) -> str:
"""GET current firmware version string."""
url = f"{API_BASE}/api/v1/devices/{DEVICE_ID}/firmware"
headers = {"Authorization": f"Bearer {token}"}
resp = client.get(url, headers=headers, timeout=TIMEOUT)
resp.raise_for_status()
data = resp.json()
# Expected shape: {"version": "12.5.3 Build 98765", "status": "up_to_date"}
return data.get("version", "").strip()
def trigger_upgrade(client: httpx.Client, token: str) -> str:
"""POST upgrade request; returns upgrade job ID."""
url = f"{API_BASE}/api/v1/devices/{DEVICE_ID}/firmware/upgrade"
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
# The API expects a payload indicating which image to use; usually null = latest.
payload = {"force": False, "schedule": None}
resp = client.post(url, json=payload, headers=headers, timeout=TIMEOUT)
resp.raise_for_status()
data = resp.json()
job_id = data.get("job_id")
if not job_id:
raise RuntimeError("Upgrade request did not return a job_id")
return job_id
def poll_upgrade_status(client: httpx.Client, token: str, job_id: str) -> bool:
"""Poll until upgrade finishes, succeeds, or times out."""
url = f"{API_BASE}/api/v1/jobs/{job_id}"
headers = {"Authorization": f"Bearer {token}"}
for attempt in tqdm(range(MAX_RETRIES), desc="Waiting for upgrade"):
resp = client.get(url, headers=headers, timeout=TIMEOUT)
resp.raise_for_status()
data = resp.json()
status = data.get("status", "").lower()
if status in ("succeeded", "completed"):
logger.success(f"Upgrade job {job_id} finished successfully.")
return True
if status in ("failed", "error", "canceled"):
logger.error(f"Upgrade job {job_id} failed: {data.get('message')}")
return False
# still running → wait
time.sleep(POLL_INTERVAL)
logger.error(f"Upgrade job {job_id} did not finish within expected time.")
return False
def main() -> None:
logger.info("Starting WatchGuard Fireware patch check…")
with httpx.Client(verify=True) as client: # set verify=False only for lab with self‑signed cert
try:
token = get_auth_token(client)
logger.debug("Obtained auth token")
except Exception as exc:
logger.exception(f"Authentication failed: {exc}")
sys.exit(1)
try:
current = fetch_current_version(client, token)
logger.info(f"Current firmware version: {current}")
except Exception as exc:
logger.exception(f"Failed to fetch current version: {exc}")
sys.exit(1)
# Simple semantic comparison – works for WatchGuard's "X.Y.Z Build NNNN" format
def normalize(v: str) -> tuple:
# Extract numeric parts: "12.5.4 Build 100123" -> (12,5,4,100123)
import re
nums = list(map(int, re.findall(r"\d+", v)))
return tuple(nums)
if normalize(current) >= normalize(PATCHED_VERSION):
logger.info("Device is already running the patched version or newer. No action needed.")
return
logger.warning(f"Device version {current} is older than patched {PATCHED_VERSION}. Initiating upgrade…")
try:
job_id = trigger_upgrade(client, token)
logger.info(f"Upgrade job submitted, ID={job_id}")
except Exception as exc:
logger.exception(f"Failed to trigger upgrade: {exc}")
sys.exit(1)
success = poll_upgrade_status(client, token, job_id)
if not success:
logger.error("Upgrade process did not succeed. Check device logs.")
sys.exit(1)
# Final verification
try:
new_ver = fetch_current_version(client, token)
logger.info(f"Post‑upgrade firmware version: {new_ver}")
if normalize(new_ver) >= normalize(PATCHED_VERSION):
logger.success("Patch verification passed.")
else:
logger.error("Patch verification failed – version still outdated.")
sys.exit(1)
except Exception as exc:
logger.exception(f"Failed to verify post‑upgrade version: {exc}")
sys.exit(1)
if __name__ == "__main__":
main()
patchFireware.ts)/**
* WatchGuard Fireware OS patch automation – TypeScript version.
*
* Prerequisites:
* npm i axios dotenv p-limit
* (dev) npm i -D typescript @types/node ts-node nodemon
*
* Create a .env file (see .env.example) with:
* WG_API_URL=https://firebox.example.com
* WG_USERNAME=admin
* WG_PASSWORD=supersecret
* WG_DEVICE_ID=1
* WG_PATCHED_VERSION=12.5.4 Build 100123
*/
import axios, { AxiosInstance } from "axios";
import * as dotenv from "dotenv";
import pLimit from "p-limit";
dotenv.config();
const API_BASE = process.env.WG_API_URL?.replace(/\/+$/, "") ?? "";
const USERNAME = process.env.WG_USERNAME ?? "";
const PASSWORD = process.env.WG_PASSWORD ?? "";
const DEVICE_ID = process.env.WG_DEVICE_ID ?? "";
const PATCHED_VERSION = process.env.WG_PATCHED_VERSION ?? "";
const TIMEOUT = Number(process.env.WG_TIMEOUT ?? "30");
const POLL_INTERVAL = Number(process.env.WG_POLL_INTERVAL ?? "15");
const MAX_RETRIES = Number(process.env.WG_MAX_RETRIES ?? "12");
if (![API_BASE, USERNAME, PASSWORD, DEVICE_ID, PATCHED_VERSION].every(Boolean)) {
console.error(
"Missing required environment variables. Check .env file (see .env.example)"
);
process.exit(1);
}
/**
* Simple HTTP client with automatic Bearer token injection.
*/
class WatchGuardClient {
private axios: AxiosInstance;
private token: string | null = null;
constructor(baseURL: string) {
this.axios = axios.create({
baseURL,
timeout: TIMEOUT * 1000,
validateStatus: (status) => status < 500, // we handle non‑2xx ourselves
});
}
/** Authenticate and store token */
async login(): Promise<void> {
try {
const { data } = await this.axios.post("/api/v1/auth/token", {
username: USERNAME,
password: PASSWORD,
});
this.token = data.access_token;
if (!this.token) throw new Error("No access_token in response");
this.axios.defaults.headers.common[
"Authorization"
] = `Bearer ${this.token}`;
} catch (err: any) {
throw new Error(`Authentication failed: ${err.response?.data?.message ?? err.message}`);
}
}
/** GET current firmware version */
async getCurrentVersion(): Promise<string> {
const { data } = await this.axios.get(`/api/v1/devices/${DEVICE_ID}/firmware`);
return data.version ?? "";
}
/** POST upgrade request */
async triggerUpgrade(): Promise<string> {
const { data } = await this.axios.post(
`/api/v1/devices/${DEVICE_ID}/firmware/upgrade`,
{ force: false, schedule: null },
{ headers: { "Content-Type": "application/json" } }
);
const jobId = data.job_id;
if (!jobId) throw new Error("Upgrade request missing job_id");
return jobId;
}
/** Poll a job until finished or timeout */
async waitForJob(jobId: string): Promise<boolean> {
const limit = pLimit(1); // serial polling
for (let i = 0; i < MAX_RETRIES; i++) {
await new Promise((res) => setTimeout(res, POLL_INTERVAL * 1000));
const { data } = await this.axios.get(`/api/v1/jobs/${jobId}`);
const status = data.status?.toLowerCase() ?? "";
if (status === "succeeded" || status === "completed") {
console.log(`✅ Job ${jobId} finished successfully.`);
return true;
}
if (status === "failed" || status === "error" || status === "canceled") {
console.error(`❌ Job ${jobId} failed: ${data.message ?? "unknown"}`);
return false;
}
// still running → continue loop
}
console.error(`⏰ Job ${jobId} did not finish within expected time.`);
return false;
}
}
/**
* Helper: turn "12.5.4 Build 100123" => [12,5,4,100123] for reliable comparison.
*/
function normalizeVersion(v: string): number[] {
return v.match(/\d+/g)?.map(Number) ?? [];
}
async function main(): Promise<void> {
const client = new WatchGuardClient(API_BASE);
try {
await client.login();
console.log("🔐 Authenticated");
} catch (e: any) {
console.error(e.message);
process.exit(1);
}
try {
const current = await client.getCurrentVersion();
console.log(`📦 Current firmware: ${current}`);
} catch (e: any) {
console.error(`❌ Failed to fetch version: ${e.message}`);
process.exit(1);
}
const curNorm = normalizeVersion(await client.getCurrentVersion());
const patchNorm = normalizeVersion(PATCHED_VERSION);
if (curNorm.length && patchNorm.length && curNorm.some((v, i) => v > patchNorm[i] ?? 0)) {
console.log("✅ Device already at or above patched version.");
return;
}
console.log(
`⚠️ Device version ${await client.getCurrentVersion()} is older than patched ${PATCHED_VERSION}. Starting upgrade…`
);
let jobId: string;
try {
jobId = await client.triggerUpgrade();
console.log(`🚀 Upgrade job submitted, ID=${jobId}`);
} catch (e: any) {
console.error(`❌ Failed to trigger upgrade: ${e.message}`);
process.exit(1);
}
const success = await client.waitForJob(jobId);
if (!success) {
process.exit(1);
}
// Final verification
try {
const finalVer = await client.getCurrentVersion();
console.log(`🔎 Post‑upgrade firmware: ${finalVer}`);
if (normalizeVersion(finalVer).every((v, i) => v >= (patchNorm[i] ?? 0))) {
console.log("🎉 Patch verification succeeded.");
} else {
console.error("💥 Patch verification failed – version still outdated.");
process.exit(1);
}
} catch (e: any) {
console.error(`❌ Verification error: ${e.message}`);
process.exit(1);
}
}
main().catch((err) => {
console.error("Unexpected error:", err);
process.exit(1);
});
Tip: To run the TS version quickly:
npx ts-node patchFireware.ts # or compile first: tsc && node patchFireware.js
Create a .env file in the project root (copy from .env.example).
Never commit real secrets to source control – add .env to .gitignore.
# .env.example
# Base URL of the WatchGuard Management API (include protocol, no trailing slash)
WG_API_URL=https://firebox.example.com
# Credentials for an API‑enabled admin user
WG_USERNAME=api_user
WG_PASSWORD=SuperSecretPassword123
# Device identifier as shown in the UI (numeric ID or UUID)
WG_DEVICE_ID=42
# The exact patched version string released by WatchGuard for the CVE
# Example from the advisory: "12.5.4 Build 100123"
WG_PATCHED_VERSION=12.5.4 Build 100123
# Optional tuning
WG_TIMEOUT=30 # seconds for each HTTP request
WG_POLL_INTERVAL=15 # seconds between job status checks
WG_MAX_RETRIES=12 # ~3 minutes total wait (12 * 15s)
Environment‑variable precedence:
.env are loaded by dotenv (Python) / dotenv (Node).export WG_API_URL=…).| Pattern | Description | Where it appears |
|---|---|---|
| Token‑based auth with automatic header injection | After login, set Authorization: Bearer <token> on the Axios/HTTPX client so every request inherits it. | WatchGuardClient.login() (TS) / get_auth_token() (PY) |
| Separation of concerns | Small, pure functions for each API call (fetch_current_version, trigger_upgrade, poll_upgrade_status). Makes unit‑testing trivial. | Python file |
| Retry/back‑off loop with timeout | Polling uses a fixed interval and a max retry count; you could easily swap in exponential back‑off (await sleep(POLL_INTERVAL * 2**i)). | Both implementations |
| Version normalisation | Extract all numeric groups (\d+) and compare as tuples to avoid lexicographic pitfalls ("12.10" > "12.2"). | normalizeVersion (TS) / normalize helper (PY) |
| Structured logging | loguru (PY) and console with emojis (TS) give clear, greppable output. In production you’d replace with a logger that outputs JSON. | Throughout |
| Graceful exit & error bubbling | Any unexpected error logs a stack trace and exits with non‑zero status – CI/CD pipelines can detect failure. | except Exception as exc: logger.exception… |
Configuration via .env | Keeps secrets out of code and enables the same binary to run against dev/stage/prod boxes. | Top of both scripts |
| Symptom | Likely Cause | Fix |
|---|---|---|
401 Unauthorized on every request | Wrong username/password, or the API user lacks the firewall_admin role. | Verify credentials in WatchGuard UI → System → Administrators. Ensure the user has API access enabled. |
SSL: CERTIFICATE_VERIFY_FAILED | Self‑signed certificate on the Firebox and verify=True (default). | For lab: set verify=False in the HTTPX/Axios client only for testing. In production, add the Firebox cert to the trusted CA bundle or use a CA‑signed cert. |
404 Not Found on /api/v1/devices/{ID}/firmware | Wrong device ID (maybe using the serial number instead of the internal numeric ID). | In the Web UI, go to Monitor → Device Manager → click the device → the URL contains deviceId=…. Use that number. |
Upgrade job stays in "running" forever then times out | The appliance cannot reach the Update Server (no internet, proxy mis‑configured, or DNS broken). | Check the Firebox’s System → Update Server settings. Ensure it can reach updates.watchguard.com on HTTPS (443). |
Upgrade request missing job_id | The firmware image you’re trying to apply is already installed or the build number is malformed. | Confirm WG_PATCHED_VERSION matches exactly what’s shown in System → Update Server → Available Updates. |
Polling loop exits early with "failed" but no message | The upgrade was rejected because the device is running a custom image or has a hardware mismatch. | Verify the device model matches the firmware build (e.g., XTM 25 vs XTM 500). Use the correct image from the WatchGuard support portal. |
| Script hangs after login | Network timeout or the API endpoint is blocked by a corporate firewall. | Test connectivity: curl -k https://<firebox>/api/v1/auth/token -d '{"username":"…","password":"…"}' -H "Content-Type: application/json" . |
Debug tip: Enable HTTP request logging.
import http.client as http_client
http_client.HTTPConnection.debuglevel = 1
const axios = require("axios");
axios.defaults.adapter = require("axios/lib/adapters/http");
axios.interceptors.request.use(req => {
console.log("REQUEST:", req.method, req.url);
return req;
});
| ✅ Item | Why it matters |
|---|---|
Use a dedicated service account with the least privileges needed (only firewall_admin + API access). | |
Store credentials in a secret manager (AWS Secrets Manager, HashiCorp Vault, Azure Key Vault) and inject them at runtime instead of a plain .env. | |
| Validate TLS certificates – never disable verification in prod. Add the Firebox’s CA‑signed cert to the system trust store. | |
| Implement idempotency – if the script runs twice, it should detect that the device is already patched and exit cleanly (already done). | |
Log to a central system (Syslog, Splunk, ELK) – replace console print/logger calls with a structured logger that outputs JSON. | |
Metric & alerting – expose a Prometheus gauge (fireware_patch_version) or send a notification (Slack, PagerDuty) when a device is out‑of‑date or upgrade fails. | |
| Test in a staging/fail‑safe cluster – run the automation against a non‑production Firebox first; verify that the upgrade does not disrupt traffic. | |
Rollback plan – know how to revert to the previous firmware version via the Web UI or CLI (set system firmware rollback) in case the new build introduces regressions. | |
Version pinning – lock the exact WG_PATCHED_VERSION variable to the build number recommended by WatchGuard for the CVE; avoid “latest” unless you have a robust canary process. | |
Audit trail – enable API request logging on the Firebox (System → Diagnostics → Log Settings → API) so you have a record of who triggered the upgrade and when. | |
| Document the runbook – include the exact steps, expected duration, and who to contact if the job fails. Store it in your internal wiki alongside the code. |
You now have:
patch_fireware.py) and a TypeScript script (patchFireware.ts) that safely checks and applies the WatchGuard Fireware OS code‑injection patch.Stay safe, keep your firewalls up‑to‑date, and remember: automation is only as good as the verification that backs it up.
Happy patching! 🚀
Source: Security Week AI
Follow ICARAX for more AI insights and tutorials.
