feat(opnsense): implement send_wake_on_lan() via the os-wol plugin API
Calls POST /api/wol/wol/set with no uuid in the payload, which makes the os-wol plugin's WolController::setAction validate and wake immediately without persisting a host to config.xml. Requires the os-wol plugin installed and the target interface to have a static IPv4 (OPNsense derives the broadcast address from the interface's own IP/subnet). Endpoint/payload verified against the plugin's source (opnsense/plugins net/wol), not guessed.
This commit is contained in:
@@ -1697,6 +1697,49 @@ class OPNsenseDriver(FirewallDriver):
|
|||||||
community = general.get("community", "public") or "public"
|
community = general.get("community", "public") or "public"
|
||||||
return SNMPConfigDict(running=True, community=community, port=161, version="2c")
|
return SNMPConfigDict(running=True, community=community, port=161, version="2c")
|
||||||
|
|
||||||
|
# ── Wake-on-LAN ──────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def send_wake_on_lan(self, mac_address: str, interface: str = "") -> dict[str, Any]:
|
||||||
|
"""Send a Wake-on-LAN magic packet via OPNsense's os-wol plugin.
|
||||||
|
|
||||||
|
Calls ``POST /api/wol/wol/set`` with an ephemeral (non-persisted) entry —
|
||||||
|
omitting ``uuid`` from the payload makes OPNsense's ``WolController::setAction``
|
||||||
|
validate and wake immediately without saving a host to config.xml. Requires
|
||||||
|
the os-wol plugin to be installed and the target interface to have a static
|
||||||
|
IPv4 address configured (OPNsense computes the broadcast address from the
|
||||||
|
interface's own IP/subnet; DHCP-assigned interfaces are rejected).
|
||||||
|
|
||||||
|
:param interface: OPNsense's *assigned* interface identifier (e.g. "lan",
|
||||||
|
"opt1") — NOT the physical device name returned by get_interfaces()/
|
||||||
|
get_networks() (e.g. "em0"). Required by this driver.
|
||||||
|
:raises ValueError: if interface is not provided.
|
||||||
|
"""
|
||||||
|
if not interface:
|
||||||
|
raise ValueError("OPNsense requires an interface for Wake-on-LAN")
|
||||||
|
try:
|
||||||
|
result = self._post(
|
||||||
|
"/api/wol/wol/set",
|
||||||
|
{"wake": {"interface": interface, "mac": mac_address.strip()}},
|
||||||
|
)
|
||||||
|
except Exception as exc:
|
||||||
|
logger.warning("Wake-on-LAN send failed for %s via %s: %s", mac_address, interface, exc)
|
||||||
|
return {"success": False, "output": str(exc)}
|
||||||
|
|
||||||
|
status = result.get("status")
|
||||||
|
if status == "error":
|
||||||
|
return {
|
||||||
|
"success": False,
|
||||||
|
"output": result.get("error_msg", "OPNsense rejected the Wake-on-LAN request"),
|
||||||
|
}
|
||||||
|
if not result:
|
||||||
|
# Model validation (malformed MAC/interface) fails silently with an
|
||||||
|
# empty {} response at HTTP 200 — no error_msg to surface.
|
||||||
|
return {
|
||||||
|
"success": False,
|
||||||
|
"output": "OPNsense rejected the request (check interface identifier and MAC format)",
|
||||||
|
}
|
||||||
|
return {"success": True, "output": f"Magic packet sent to {mac_address} via {interface}"}
|
||||||
|
|
||||||
def run_device_action(self, action: str) -> dict[str, Any]:
|
def run_device_action(self, action: str) -> dict[str, Any]:
|
||||||
"""Execute a named action on the firewall."""
|
"""Execute a named action on the firewall."""
|
||||||
if action == "fix_snmp":
|
if action == "fix_snmp":
|
||||||
|
|||||||
@@ -1007,6 +1007,54 @@ class TestInternalPost:
|
|||||||
assert result == {"result": "saved"}
|
assert result == {"result": "saved"}
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# send_wake_on_lan()
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
class TestSendWakeOnLan:
|
||||||
|
def test_raises_value_error_without_interface(self, driver):
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF")
|
||||||
|
driver.session.post.assert_not_called()
|
||||||
|
|
||||||
|
def test_sends_correct_request(self, driver):
|
||||||
|
driver.session.post.return_value = _make_json_response({"status": ""})
|
||||||
|
driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
|
||||||
|
driver.session.post.assert_called_once_with(
|
||||||
|
"https://opnsense.example.com/api/wol/wol/set",
|
||||||
|
json={"wake": {"interface": "lan", "mac": "AA:BB:CC:DD:EE:FF"}},
|
||||||
|
timeout=60,
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_success_on_empty_status(self, driver):
|
||||||
|
# OPNsense's WolController returns {"status": trim(configd output)} on
|
||||||
|
# success — the underlying `wol` CLI tool typically prints nothing.
|
||||||
|
driver.session.post.return_value = _make_json_response({"status": ""})
|
||||||
|
result = driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
|
||||||
|
assert result == {"success": True, "output": "Magic packet sent to AA:BB:CC:DD:EE:FF via lan"}
|
||||||
|
|
||||||
|
def test_returns_failure_on_error_status(self, driver):
|
||||||
|
driver.session.post.return_value = _make_json_response(
|
||||||
|
{"status": "error", "error_msg": "Incorrect IPv4 configuration on interface"}
|
||||||
|
)
|
||||||
|
result = driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="opt1")
|
||||||
|
assert result == {"success": False, "output": "Incorrect IPv4 configuration on interface"}
|
||||||
|
|
||||||
|
def test_returns_failure_on_empty_response(self, driver):
|
||||||
|
# OPNsense's model validation (bad MAC/interface) silently returns {}
|
||||||
|
# with HTTP 200 rather than an error status.
|
||||||
|
driver.session.post.return_value = _make_json_response({})
|
||||||
|
result = driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
|
||||||
|
assert result["success"] is False
|
||||||
|
|
||||||
|
def test_returns_failure_on_request_exception(self, driver):
|
||||||
|
# e.g. HTTP 404 — the os-wol plugin is not installed on this firewall.
|
||||||
|
driver.session.post.side_effect = Exception("404 Client Error: Not Found")
|
||||||
|
result = driver.send_wake_on_lan("AA:BB:CC:DD:EE:FF", interface="lan")
|
||||||
|
assert result["success"] is False
|
||||||
|
assert "404" in result["output"]
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# load_merge_candidate()
|
# load_merge_candidate()
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|||||||
Reference in New Issue
Block a user