"""Public OPNsense client class composed from domain mixins."""
from types import TracebackType
from typing import Self
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 .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
[docs]
class OPNsenseClient(
ClientBaseMixin,
FirmwareMixin,
FirewallMixin,
DHCPMixin,
ServicesMixin,
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()
[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.
"""
orig_throw_errors = self._throw_errors
self._throw_errors = True
try:
try:
fw_ver = await self.get_host_firmware_version()
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
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
finally:
self._throw_errors = orig_throw_errors