# This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, version 2 of the License. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # vim:sw=4:ts=4:et import asyncio import contextlib from functools import lru_cache import logging import re import socket import ssl import threading from contextlib import asynccontextmanager from typing import AsyncIterator, Optional, Tuple, Union import httpx import requests from requests.adapters import HTTPAdapter from urllib3.connection import HTTPConnection from .config import ( KEEPALIVE_COUNT, KEEPALIVE_IDLE, KEEPALIVE_INTERVAL, MAX_RETRY_HTTP_CONNECTION, TIMEOUT_HTTP_PROTOCOL, ) from .exceptions import CommError, LoginError, ReadTimeoutError from .utils import clean_url, pretty _LOGGER = logging.getLogger(__name__) _KEEPALIVE_OPTS = HTTPConnection.default_socket_options + [ (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1), ] # On some systems TCP_KEEP* are not defined in socket. try: _KEEPALIVE_OPTS += [ (socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, KEEPALIVE_IDLE), (socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, KEEPALIVE_INTERVAL), (socket.IPPROTO_TCP, socket.TCP_KEEPCNT, KEEPALIVE_COUNT), ] except AttributeError: pass TimeoutT = Union[Optional[float], Tuple[Optional[float], Optional[float]]] HttpxTimeoutT = Union[ Optional[float], Tuple[Optional[float], Optional[float], Optional[float], Optional[float]], ] @lru_cache(maxsize=None) def create_default_ssl_context() -> ssl.SSLContext: """Return an SSL context that verifies the server certificate.""" return ssl.create_default_context() @lru_cache(maxsize=None) def create_no_verify_ssl_context() -> ssl.SSLContext: """Return an SSL context that does not verify the server certificate. This is a copy of aiohttp's create_default_context() function, with the ssl verify turned off and old SSL versions enabled. https://github.com/aio-libs/aiohttp/blob/33953f110e97eecc707e1402daa8d543f38a189b/aiohttp/connector.py#L911 """ sslcontext = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) sslcontext.check_hostname = False sslcontext.verify_mode = ssl.CERT_NONE # Allow all ciphers rather than only Python 3.10 default sslcontext.set_ciphers("DEFAULT") with contextlib.suppress(AttributeError): # This only works for OpenSSL >= 1.0.0 sslcontext.options |= ssl.OP_NO_COMPRESSION sslcontext.set_default_verify_paths() # ssl.OP_LEGACY_SERVER_CONNECT is only available in Python 3.12a4+ sslcontext.options |= getattr(ssl, "OP_LEGACY_SERVER_CONNECT", 0x4) return sslcontext class SOHTTPAdapter(HTTPAdapter): """HTTPAdapter with support for socket options.""" def __init__(self, *args, **kwargs) -> None: self.socket_options = kwargs.pop("socket_options", None) super().__init__(*args, **kwargs) def init_poolmanager(self, *args, **kwargs) -> None: if self.socket_options is not None: kwargs["socket_options"] = self.socket_options super().init_poolmanager(*args, **kwargs) class Http: def __init__( self, host: str, port: int, user: str, password: str, *, verbose: bool = True, protocol: str = "http", ssl_verify: bool = True, retries_connection: Optional[int] = None, timeout_protocol: TimeoutT = None, ) -> None: self._token_lock = threading.Lock() try: self._async_token_lock: Optional[asyncio.Lock] = asyncio.Lock() except RuntimeError: self._async_token_lock = None self._cmd_id_lock = threading.Lock() self._cmd_id = 0 self._host = clean_url(host) self._port = port self._user = user self._password = password self._verbose = verbose self._protocol = protocol self._verify = ssl_verify self._base_url = self.__base_url() self._retries_default = ( retries_connection if retries_connection is not None else MAX_RETRY_HTTP_CONNECTION ) self._timeout_default = timeout_protocol or TIMEOUT_HTTP_PROTOCOL self._token: Optional[requests.auth.AuthBase] = None self._async_token: Optional[httpx.Auth] = None self._name: Optional[str] = None self._serial: Optional[str] = None def _generate_token(self) -> None: """Create authentation to use with requests.""" cmd = "magicBox.cgi?action=getMachineName" _LOGGER.debug("%s Trying Basic Authentication", self) self._token = requests.auth.HTTPBasicAuth(self._user, self._password) try: try: resp = self._command(cmd).content.decode() except LoginError: _LOGGER.debug("%s Trying Digest Authentication", self) self._token = requests.auth.HTTPDigestAuth( self._user, self._password ) resp = self._command(cmd).content.decode() except CommError: self._token = None raise # check if user passed result = resp.lower() if "invalid" in result or "error" in result: _LOGGER.debug( "%s Result from camera: %s", self, resp.strip().replace("\r\n", ": "), ) self._token = None raise LoginError("Invalid credentials") self._name = pretty(resp.strip()) _LOGGER.debug("%s Retrieving serial number", self) self._serial = pretty( self._command("magicBox.cgi?action=getSerialNo") .content.decode() .strip() ) async def _async_generate_token(self) -> None: """Create authentation to use with requests.""" cmd = "magicBox.cgi?action=getMachineName" _LOGGER.debug("%s Trying async Basic Authentication", self) self._async_token = httpx.BasicAuth(self._user, self._password) try: try: resp = (await self._async_command(cmd)).content.decode() except LoginError: _LOGGER.debug("%s Trying async Digest Authentication", self) self._async_token = httpx.DigestAuth( self._user, self._password ) resp = (await self._async_command(cmd)).content.decode() except CommError: self._async_token = None raise # check if user passed result = resp.lower() if "invalid" in result or "error" in result: _LOGGER.debug( "%s Result from camera: %s", self, resp.strip().replace("\r\n", ": "), ) self._async_token = None raise LoginError("Invalid credentials") self._name = pretty(resp.strip()) _LOGGER.debug("%s Retrieving serial number", self) self._serial = pretty( (await self._async_command("magicBox.cgi?action=getSerialNo")) .content.decode() .strip() ) def __repr__(self) -> str: """Default object representation.""" if self._name is None: return f"" if self._serial is None: return f"<{self._name}:CONNECTING>" return f"<{self._name}:{self._serial}>" def as_dict(self) -> dict: """Callback for __dict__.""" cdict = self.__dict__.copy() redacted = "**********" if cdict["_token"] is not None: cdict["_token"] = redacted cdict["_password"] = redacted return cdict # Base methods def __base_url(self, param: str = "") -> str: return f"{self._protocol}://{self._host}:{self._port}/cgi-bin/{param}" def get_base_url(self) -> str: return self._base_url def command(self, *args, **kwargs) -> requests.Response: """ Args: cmd - command to execute via http retries - maximum number of retries each connection should attempt timeout_cmd - timeout stream - if True do not download entire response immediately """ with self._token_lock: if not self._token: self._generate_token() return self._command(*args, **kwargs) async def async_command(self, *args, **kwargs) -> httpx.Response: if self._async_token_lock is None: raise RuntimeError("Camera not setup with async loop running") async with self._async_token_lock: if not self._async_token: await self._async_generate_token() return await self._async_command(*args, **kwargs) @asynccontextmanager async def async_stream_command( self, *args, **kwargs ) -> AsyncIterator[httpx.Response]: if self._async_token_lock is None: raise RuntimeError("Camera not setup with async loop running") async with self._async_token_lock: if not self._async_token: await self._async_generate_token() async with self._async_stream_command(*args, **kwargs) as ret: yield ret def _command( self, cmd: str, retries: Optional[int] = None, timeout_cmd: TimeoutT = None, stream: bool = False, ) -> requests.Response: url = self.__base_url(cmd) with self._cmd_id_lock: cmd_id = self._cmd_id = self._cmd_id + 1 _LOGGER.debug("%s HTTP query %i: %s", self, cmd_id, url) if retries is None: retries = self._retries_default timeout = timeout_cmd or self._timeout_default with requests.Session() as session: if isinstance(timeout, float): use_keepalive = False else: use_keepalive = timeout[0] is None or timeout[1] is None if use_keepalive: session.mount( "{0}://".format(self._protocol), SOHTTPAdapter(socket_options=_KEEPALIVE_OPTS), ) for loop in range(1, 2 + retries): _LOGGER.debug( "%s Running query %i attempt %s", self, cmd_id, loop ) try: resp = session.get( url, auth=self._token, stream=stream, timeout=timeout, verify=self._verify, ) if resp.status_code == 401: _LOGGER.debug( "%s Query %i: Unauthorized (401)", self, cmd_id ) self._token = None raise LoginError() resp.raise_for_status() except requests.RequestException as error: _LOGGER.debug( "%s Query %i failed due to error: %r", self, cmd_id, error, ) if loop > retries: raise CommError(error) from error msg = re.sub( r"at 0x[0-9a-fA-F]+", "at ADDRESS", repr(error) ) _LOGGER.warning( "%s Trying again due to error: %s", self, msg ) continue else: break _LOGGER.debug( "%s Query %i worked. Exit code: <%s>", self, cmd_id, resp.status_code, ) return resp async def _async_command( self, cmd: str, retries: Optional[int] = None, timeout_cmd: TimeoutT = None, ) -> httpx.Response: url = self.__base_url(cmd) cmd_id = self._cmd_id = self._cmd_id + 1 _LOGGER.debug("%s Async HTTP query %i: %s", self, cmd_id, url) if retries is None: retries = self._retries_default timeout = timeout_cmd or self._timeout_default # requests uses (connect timeout, read timeout), and httpx adds (write # timeout, pool timeout), just set the extras to None if isinstance(timeout, tuple): httpx_timeout: HttpxTimeoutT = timeout + (None, None) else: httpx_timeout = timeout if self._verify: ssl_context = create_default_ssl_context() else: ssl_context = create_no_verify_ssl_context() async with httpx.AsyncClient( follow_redirects=True, auth=self._async_token, verify=ssl_context, timeout=httpx_timeout, ) as client: for loop in range(1, 2 + retries): _LOGGER.debug( "%s Running query %i attempt %s", self, cmd_id, loop ) try: resp = await client.get(url) if resp.status_code == 401: _LOGGER.debug( "%s Query %i: Unauthorized (401)", self, cmd_id ) self._async_token = None raise LoginError() resp.raise_for_status() except httpx.HTTPError as error: _LOGGER.debug( "%s Query %i failed due to error: %r", self, cmd_id, error, ) if loop > retries: raise CommError(error) from error msg = re.sub( r"at 0x[0-9a-fA-F]+", "at ADDRESS", repr(error) ) _LOGGER.warning( "%s Trying again due to error: %s", self, msg ) continue else: break _LOGGER.debug( "%s Query %i worked. Exit code: <%s>", self, cmd_id, resp.status_code, ) return resp @asynccontextmanager async def _async_stream_command( self, cmd: str, timeout_cmd: TimeoutT = None, ) -> AsyncIterator[httpx.Response]: url = self.__base_url(cmd) cmd_id = self._cmd_id = self._cmd_id + 1 _LOGGER.debug("%s HTTP query %i: %s", self, cmd_id, url) timeout = timeout_cmd or self._timeout_default # requests uses (connect timeout, read timeout), and httpx adds (write # timeout, pool timeout), just set the extras to None if isinstance(timeout, tuple): httpx_timeout: HttpxTimeoutT = timeout + (None, None) else: httpx_timeout = timeout async with httpx.AsyncClient( follow_redirects=True, auth=self._async_token, verify=self._verify, timeout=httpx_timeout, ) as client: try: async with client.stream("GET", url) as resp: if resp.status_code == 401: _LOGGER.debug( "%s Query %i: Unauthorized (401)", self, cmd_id ) self._async_token = None raise LoginError() resp.raise_for_status() _LOGGER.debug( "%s Query %i worked. Exit code: <%s>", self, cmd_id, resp.status_code, ) yield resp except httpx.HTTPError as error: _LOGGER.debug( "%s Query %i failed due to error: %r", self, cmd_id, error, ) if isinstance(error, httpx.ReadTimeout): raise ReadTimeoutError(error) from error else: raise CommError(error) from error def command_audio( self, cmd: str, file_content, http_header, timeout: TimeoutT = None, ) -> None: with self._token_lock: if not self._token: self._generate_token() url = self.__base_url(cmd) try: requests.post( url, files=file_content, auth=self._token, headers=http_header, timeout=timeout or self._timeout_default, ) except requests.exceptions.ReadTimeout: pass # Helpers for common commands def _get_config(self, config_name: str) -> str: ret = self.command( f"configManager.cgi?action=getConfig&name={config_name}" ) return ret.content.decode() async def _async_get_config(self, config_name: str) -> str: ret = await self.async_command( f"configManager.cgi?action=getConfig&name={config_name}" ) return ret.content.decode() def _magic_box(self, action: str) -> str: ret = self.command(f"magicBox.cgi?action={action}") return ret.content.decode() async def _async_magic_box(self, action: str) -> str: ret = await self.async_command(f"magicBox.cgi?action={action}") return ret.content.decode()