155 lines
4.9 KiB
Python
155 lines
4.9 KiB
Python
"""AdGuard Home API client."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import Any
|
|
|
|
import aiohttp
|
|
import async_timeout
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
|
|
from .const import (
|
|
API_CLIENTS,
|
|
API_DNS_CONFIG,
|
|
API_STATUS,
|
|
API_STATS,
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class AdGuardHomeApiError(Exception):
|
|
"""Exception to indicate a general API error."""
|
|
|
|
|
|
class AdGuardHomeConnectionError(AdGuardHomeApiError):
|
|
"""Exception to indicate a connection error."""
|
|
|
|
|
|
class AdGuardHomeAuthError(AdGuardHomeApiError):
|
|
"""Exception to indicate an authentication error."""
|
|
|
|
|
|
class AdGuardHomeAPI:
|
|
"""AdGuard Home API client."""
|
|
|
|
def __init__(
|
|
self,
|
|
session: aiohttp.ClientSession,
|
|
host: str,
|
|
port: int,
|
|
username: str | None = None,
|
|
password: str | None = None,
|
|
ssl: bool = False,
|
|
verify_ssl: bool = True,
|
|
) -> None:
|
|
"""Initialize the API client."""
|
|
self._session = session
|
|
self._host = host
|
|
self._port = port
|
|
self._username = username
|
|
self._password = password
|
|
self._ssl = ssl
|
|
self._verify_ssl = verify_ssl
|
|
|
|
protocol = "https" if ssl else "http"
|
|
self._base_url = f"{protocol}://{host}:{port}"
|
|
|
|
async def _request(
|
|
self,
|
|
method: str,
|
|
endpoint: str,
|
|
data: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Make a request to the AdGuard Home API."""
|
|
url = f"{self._base_url}{endpoint}"
|
|
auth = None
|
|
|
|
if self._username and self._password:
|
|
auth = aiohttp.BasicAuth(self._username, self._password)
|
|
|
|
headers = {"Content-Type": "application/json"}
|
|
|
|
try:
|
|
async with async_timeout.timeout(10):
|
|
async with self._session.request(
|
|
method,
|
|
url,
|
|
json=data,
|
|
auth=auth,
|
|
headers=headers,
|
|
ssl=self._verify_ssl,
|
|
) as response:
|
|
if response.status == 401:
|
|
raise AdGuardHomeAuthError("Authentication failed")
|
|
|
|
if response.status == 403:
|
|
raise AdGuardHomeAuthError("Access forbidden")
|
|
|
|
if response.status not in (200, 204):
|
|
text = await response.text()
|
|
raise AdGuardHomeApiError(
|
|
f"Request failed with status {response.status}: {text}"
|
|
)
|
|
|
|
if response.status == 204:
|
|
return {}
|
|
|
|
return await response.json()
|
|
|
|
except asyncio.TimeoutError as err:
|
|
raise AdGuardHomeConnectionError("Timeout connecting to AdGuard Home") from err
|
|
except aiohttp.ClientError as err:
|
|
raise AdGuardHomeConnectionError(f"Error connecting to AdGuard Home: {err}") from err
|
|
|
|
async def get_status(self) -> dict[str, Any]:
|
|
"""Get AdGuard Home status."""
|
|
return await self._request("GET", API_STATUS)
|
|
|
|
async def get_stats(self) -> dict[str, Any]:
|
|
"""Get AdGuard Home statistics."""
|
|
return await self._request("GET", API_STATS)
|
|
|
|
async def get_clients(self) -> dict[str, Any]:
|
|
"""Get AdGuard Home clients."""
|
|
return await self._request("GET", API_CLIENTS)
|
|
|
|
async def set_protection(self, enabled: bool) -> None:
|
|
"""Enable or disable protection."""
|
|
data = {"protection_enabled": enabled}
|
|
await self._request("POST", API_DNS_CONFIG, data)
|
|
|
|
async def set_filtering(self, enabled: bool) -> None:
|
|
"""Enable or disable filtering."""
|
|
data = {"filtering_enabled": enabled}
|
|
await self._request("POST", API_DNS_CONFIG, data)
|
|
|
|
async def set_safebrowsing(self, enabled: bool) -> None:
|
|
"""Enable or disable safe browsing."""
|
|
data = {"safebrowsing_enabled": enabled}
|
|
await self._request("POST", API_DNS_CONFIG, data)
|
|
|
|
async def set_parental_control(self, enabled: bool) -> None:
|
|
"""Enable or disable parental control."""
|
|
data = {"parental_enabled": enabled}
|
|
await self._request("POST", API_DNS_CONFIG, data)
|
|
|
|
async def set_safe_search(self, enabled: bool) -> None:
|
|
"""Enable or disable safe search."""
|
|
data = {"safesearch_enabled": enabled}
|
|
await self._request("POST", API_DNS_CONFIG, data)
|
|
|
|
async def set_query_log(self, enabled: bool) -> None:
|
|
"""Enable or disable query log."""
|
|
data = {"querylog_enabled": enabled}
|
|
await self._request("POST", API_DNS_CONFIG, data)
|
|
|
|
async def test_connection(self) -> bool:
|
|
"""Test connection to AdGuard Home."""
|
|
try:
|
|
await self.get_status()
|
|
return True
|
|
except AdGuardHomeApiError:
|
|
return False
|