chore: adopt the NAPALM driver standard #31
No files matched your search
@@ -65,7 +65,8 @@ def _service_of(command: str, user: str, services: set[str]) -> str | None:
|
||||
def _address(protocol: str, address: str) -> tuple:
|
||||
"""``(address, interface)``: ``*`` is every address of the socket's family."""
|
||||
if address == "*":
|
||||
return ("::" if protocol.endswith("6") else "0.0.0.0"), None
|
||||
# A reported wildcard address (data from sockstat), not a bind.
|
||||
return ("::" if protocol.endswith("6") else "0.0.0.0"), None # nosec B104
|
||||
host, _, zone = address.partition("%")
|
||||
return host, zone or None
|
||||
|
||||
|
||||
@@ -44,12 +44,6 @@ import time
|
||||
from ipaddress import ip_address, ip_network
|
||||
from typing import Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: How long refresh_available_updates waits for the firewall's update check.
|
||||
_REFRESH_POLLS = 20
|
||||
_REFRESH_INTERVAL = 3
|
||||
|
||||
import requests
|
||||
from napalm.base.exceptions import (
|
||||
ConnectionClosedException,
|
||||
@@ -76,6 +70,12 @@ from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interf
|
||||
from napalm_opnsense.resolver_mixin import OPNsenseResolverMixin
|
||||
from napalm_opnsense.virtual_ips import carp_addresses
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: How long refresh_available_updates waits for the firewall's update check.
|
||||
_REFRESH_POLLS = 20
|
||||
_REFRESH_INTERVAL = 3
|
||||
|
||||
_ARCHITECTURE_SUFFIX = re.compile(r"-(amd64|aarch64|arm64|i386|armv7)$")
|
||||
|
||||
|
||||
@@ -636,7 +636,8 @@ class OPNsenseDriver(
|
||||
# Clean up BSD link-layer gateway references
|
||||
clean_gateway = (
|
||||
""
|
||||
if (not gateway or gateway.startswith("link#") or gateway == "0.0.0.0")
|
||||
# A route's reported gateway compared with the wildcard, not a bind.
|
||||
if (not gateway or gateway.startswith("link#") or gateway == "0.0.0.0") # nosec B104
|
||||
else gateway
|
||||
)
|
||||
|
||||
@@ -2261,7 +2262,9 @@ class OPNsenseDriver(
|
||||
return {"success": True, "output": status.get("status_msg", "")}
|
||||
return {
|
||||
"success": False,
|
||||
"output": f"The update check did not finish within {_REFRESH_POLLS * _REFRESH_INTERVAL} s",
|
||||
"output": (
|
||||
f"The update check did not finish within {_REFRESH_POLLS * _REFRESH_INTERVAL} s"
|
||||
),
|
||||
}
|
||||
|
||||
def reboot_host(self) -> None:
|
||||
@@ -2464,7 +2467,9 @@ class OPNsenseDriver(
|
||||
# empty {} response at HTTP 200 — no error_msg to surface.
|
||||
return {
|
||||
"success": False,
|
||||
"output": "OPNsense rejected the request (check interface identifier and MAC format)",
|
||||
"output": (
|
||||
"OPNsense rejected the request (check interface identifier and MAC format)"
|
||||
),
|
||||
}
|
||||
return {"success": True, "output": f"Magic packet sent to {mac_address} via {interface}"}
|
||||
|
||||
|
||||
Reference in new issue
Block a user