"""Public OPNsense client class composed from domain mixins."""
from collections.abc import Awaitable, Callable
from types import TracebackType
from typing import Self, TypeVar
import aiohttp
import awesomeversion
from .client_base import ClientBaseMixin
from .const import OPNSENSE_LTD_FIRMWARE, OPNSENSE_MIN_FIRMWARE
from .dhcp import DHCPMixin
from .exceptions import (
OPNsenseBelowMinFirmware,
OPNsenseConnectionError,
OPNsenseInvalidAuth,
OPNsenseInvalidURL,
OPNsensePrivilegeMissing,
OPNsenseSSLError,
OPNsenseTimeoutError,
OPNsenseUnknownFirmware,
)
from .firewall import FirewallMixin
from .firmware import FirmwareMixin
from .helpers import _LOGGER
from .services import ServicesMixin
from .smart import SmartMixin
from .speedtest import SpeedtestMixin
from .system import SystemMixin
from .telemetry import TelemetryMixin
from .unbound import UnboundMixin
from .vnstat import VnstatMixin
from .vouchers import VouchersMixin
from .vpn import VPNMixin
_T = TypeVar("_T")
[docs]
class OPNsenseClient(
ClientBaseMixin,
FirmwareMixin,
FirewallMixin,
DHCPMixin,
ServicesMixin,
SmartMixin,
SpeedtestMixin,
SystemMixin,
UnboundMixin,
VouchersMixin,
TelemetryMixin,
VnstatMixin,
VPNMixin,
):
"""Async client for supported OPNsense REST endpoints."""
async def __aenter__(self) -> Self:
"""Validate the client before entering an async context manager.
Returns:
Self: Validated client instance.
"""
await self.validate()
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
traceback: TracebackType | None,
) -> None:
"""Close background resources when leaving an async context manager.
Args:
exc_type (type[BaseException] | None): Exception type raised in the context block.
exc (BaseException | None): Exception instance raised in the context block.
traceback (TracebackType | None): Traceback for an exception raised in the context block.
"""
del exc_type, exc, traceback
await self.async_close()
async def _run_validation_request(self, request: Callable[[], Awaitable[_T]]) -> _T:
"""Run a validation request and map transport failures to public exceptions.
Args:
request (Callable[[], Awaitable[_T]]): Zero-argument async request
used during client validation.
Returns:
_T: Result returned by the validation request.
Raises:
OPNsenseInvalidURL: Raised when the configured URL is invalid.
OPNsenseSSLError: Raised when the TLS handshake fails.
OPNsenseTimeoutError: Raised when validation requests time out.
OPNsenseInvalidAuth: Raised when API authentication fails.
OPNsensePrivilegeMissing: Raised when the API user lacks privileges.
OPNsenseConnectionError: Raised when another client connection error occurs.
"""
try:
return await request()
except (aiohttp.ClientConnectorDNSError, aiohttp.InvalidURL) as e:
raise OPNsenseInvalidURL from e
except aiohttp.ClientSSLError as e:
raise OPNsenseSSLError from e
except (TimeoutError, aiohttp.ServerTimeoutError) as e:
raise OPNsenseTimeoutError from e
except aiohttp.ClientResponseError as e:
if e.status == 401:
raise OPNsenseInvalidAuth from e
if e.status == 403:
raise OPNsensePrivilegeMissing from e
raise OPNsenseConnectionError from e
except aiohttp.ClientError as e:
raise OPNsenseConnectionError from e
[docs]
async def validate(self) -> None:
"""Validate connectivity, authentication, and minimum firmware support.
Note that this will throw errors, regardless of what self._throw_errors is set to.
Raises:
OPNsenseInvalidURL: Raised when the configured URL is invalid.
OPNsenseSSLError: Raised when the TLS handshake fails.
OPNsenseTimeoutError: Raised when validation requests time out.
OPNsenseInvalidAuth: Raised when API authentication fails.
OPNsensePrivilegeMissing: Raised when the API user lacks privileges.
OPNsenseConnectionError: Raised when another client connection error occurs.
OPNsenseUnknownFirmware: Raised when firmware detection returns no version.
OPNsenseBelowMinFirmware: Raised when the detected firmware is unsupported.
OPNsenseMissingDeviceUniqueID: Raised when no device unique ID can
be resolved from OPNsense.
"""
orig_throw_errors = self._throw_errors
self._throw_errors = True
try:
fw_ver = await self._run_validation_request(self.get_host_firmware_version)
if fw_ver is None:
raise OPNsenseUnknownFirmware
try:
if awesomeversion.AwesomeVersion(fw_ver) < awesomeversion.AwesomeVersion(
OPNSENSE_MIN_FIRMWARE
):
msg = (
f"OPNsense Firmware {fw_ver} detected. "
f"aiopnsense requires OPNsense Firmware >= {OPNSENSE_MIN_FIRMWARE}"
)
raise OPNsenseBelowMinFirmware(msg)
if awesomeversion.AwesomeVersion(fw_ver) < awesomeversion.AwesomeVersion(
OPNSENSE_LTD_FIRMWARE
):
_LOGGER.warning(
"OPNsense Firmware of %s is below the recommended >= %s. aiopnsense will work, but there may be some missing features.",
fw_ver,
OPNSENSE_LTD_FIRMWARE,
)
except (
awesomeversion.exceptions.AwesomeVersionCompareException,
TypeError,
ValueError,
) as err:
raise OPNsenseUnknownFirmware from err
await self._run_validation_request(self.get_device_unique_id)
finally:
self._throw_errors = orig_throw_errors