diff --git a/README.md b/README.md index 2e3abf1..317e0f4 100644 --- a/README.md +++ b/README.md @@ -239,11 +239,11 @@ Remove passwords and private addresses before posting logs or XML. ### Media file is too large -Remote media downloads are limited to 20 MB. Use short announcement files where possible. +Media downloads are limited to 20 MB. Use short announcement files where possible. ### HTTPS devices -The current implementation uses raw HTTP sockets to the Hikvision ISAPI port and is designed for local-network HTTP access. HTTPS support is not implemented. +The integration currently connects to the Hikvision ISAPI port over local-network HTTP. HTTPS configuration is not implemented. ## Security Notes diff --git a/custom_components/hikvision_player/audio.py b/custom_components/hikvision_player/audio.py index 3af9449..f0ac91b 100644 --- a/custom_components/hikvision_player/audio.py +++ b/custom_components/hikvision_player/audio.py @@ -2,7 +2,6 @@ from __future__ import annotations -from pathlib import Path from typing import Iterable import miniaudio @@ -15,17 +14,6 @@ _SEG_UEND = (0x3F, 0x7F, 0xFF, 0x1FF, 0x3FF, 0x7FF, 0xFFF, 0x1FFF) -def decode_media_file(path: Path, codec: str) -> bytes: - """Decode a media file to camera-ready G.711 bytes.""" - decoded = miniaudio.decode_file( - str(path), - output_format=miniaudio.SampleFormat.SIGNED16, - nchannels=1, - sample_rate=SAMPLE_RATE, - ) - return encode_g711(decoded.samples, codec) - - def decode_media_bytes(data: bytes, codec: str) -> bytes: """Decode media bytes to camera-ready G.711 bytes.""" decoded = miniaudio.decode( diff --git a/custom_components/hikvision_player/client.py b/custom_components/hikvision_player/client.py index 69b1d55..983e254 100644 --- a/custom_components/hikvision_player/client.py +++ b/custom_components/hikvision_player/client.py @@ -4,37 +4,35 @@ import asyncio from dataclasses import dataclass -import hashlib import logging -from pathlib import Path -import re -import secrets import time -import urllib.parse import xml.etree.ElementTree as ET -from aiohttp import ClientError, ClientTimeout +from aiohttp import ( + ClientError, + ClientResponse, + ClientSession, + ClientTimeout, + DigestAuthMiddleware, +) +from aiohttp.abc import AbstractStreamWriter +from aiohttp.payload import BytesPayload, Payload +from yarl import URL from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.aiohttp_client import async_get_clientsession -from .audio import SAMPLE_RATE, decode_media_bytes, decode_media_file +from .audio import SAMPLE_RATE, decode_media_bytes _LOGGER = logging.getLogger(__name__) -CHUNK_BYTES = 160 +AUDIO_CHUNK_BYTES = 160 +DOWNLOAD_CHUNK_BYTES = 64 * 1024 MAX_MEDIA_BYTES = 20 * 1024 * 1024 - -@dataclass(frozen=True, slots=True) -class IsapiSource: - """Connection settings for a Hikvision ISAPI endpoint.""" - - host: str - port: int - username: str - password: str +DEVICE_TIMEOUT = ClientTimeout(total=10) +DOWNLOAD_TIMEOUT = ClientTimeout(total=30) @dataclass(frozen=True, slots=True) @@ -45,6 +43,45 @@ class TalkbackChannel: codec: str +class RealtimeAudioPayload(BytesPayload): + """An aiohttp payload that sends G.711 audio at its playback rate.""" + + def __init__(self, data: bytes) -> None: + """Initialize a reusable, fixed-length payload.""" + super().__init__(data, content_type="application/octet-stream") + + async def write(self, writer: AbstractStreamWriter) -> None: + """Write one-byte-per-sample G.711 audio at 8 kHz.""" + await self._async_write(writer, self._value) + + async def write_with_length( + self, + writer: AbstractStreamWriter, + content_length: int | None, + ) -> None: + """Write a length-limited payload at the audio playback rate.""" + data = self._value if content_length is None else self._value[:content_length] + await self._async_write(writer, data) + + async def _async_write( + self, + writer: AbstractStreamWriter, + data: bytes, + ) -> None: + """Write payload bytes at the audio playback rate.""" + started = time.monotonic() + sent = 0 + + for offset in range(0, len(data), AUDIO_CHUNK_BYTES): + chunk = data[offset : offset + AUDIO_CHUNK_BYTES] + await writer.write(chunk) + + sent += len(chunk) + delay = started + sent / SAMPLE_RATE - time.monotonic() + if delay > 0: + await asyncio.sleep(delay) + + class HikvisionPlayerClient: """Play media through a Hikvision ISAPI two-way audio channel.""" @@ -58,7 +95,10 @@ def __init__( ) -> None: """Initialize the client.""" self._hass = hass - self._source = IsapiSource(host, port, username, password) + self._session: ClientSession = async_get_clientsession(hass) + self._base_url = URL.build(scheme="http", host=host, port=port) + self._address = f"{host}:{port}" + self._digest_auth = DigestAuthMiddleware(username, password) self._channel: TalkbackChannel | None = None self._task: asyncio.Task[None] | None = None self._lock = asyncio.Lock() @@ -95,89 +135,74 @@ async def async_stop(self) -> None: async def _async_play_once(self, media_url: str) -> None: """Decode media, open talkback, stream audio, then close talkback.""" - source = self._source - channel = await self._async_channel(source) + channel = await self._async_channel() payload = await self._async_decode(media_url, channel.codec) - base_path = f"/ISAPI/System/TwoWayAudio/channels/{channel.channel_id}" - writer: asyncio.StreamWriter | None = None try: - await self._async_request(source, "PUT", f"{base_path}/close") - await self._async_request(source, "PUT", f"{base_path}/open") - writer = await self._async_audio_writer(source, f"{base_path}/audioData") - await _stream_realtime(writer, payload) + await self._async_device_request("PUT", f"{base_path}/close") + await self._async_device_request("PUT", f"{base_path}/open") + await self._async_device_request( + "PUT", + f"{base_path}/audioData", + data=RealtimeAudioPayload(payload), + read_body=False, + timeout=ClientTimeout( + total=max(10, len(payload) / SAMPLE_RATE + 10), + sock_connect=5, + ), + ) finally: - if writer is not None: - writer.close() - try: - await writer.wait_closed() - except OSError: - pass - try: - await self._async_request(source, "PUT", f"{base_path}/close") + await self._async_device_request("PUT", f"{base_path}/close") except HomeAssistantError as err: _LOGGER.debug("Failed to close Hikvision talkback channel: %s", err) async def _async_decode(self, media_url: str, codec: str) -> bytes: - """Download/read media and encode it as the camera G.711 payload.""" - parsed = urllib.parse.urlsplit(media_url) - if parsed.scheme in {"http", "https"}: - data = await self._async_download(media_url) - return await self._hass.async_add_executor_job( - decode_media_bytes, - data, - codec, - ) - - path = Path(media_url) - if not path.is_file(): - raise HomeAssistantError(f"Media file does not exist: {media_url}") - - return await self._hass.async_add_executor_job(decode_media_file, path, codec) + """Download media and encode it as the camera G.711 payload.""" + if URL(media_url).scheme not in {"http", "https"}: + raise HomeAssistantError(f"Unsupported media URL: {media_url}") + + data = await self._async_download(media_url) + return await self._hass.async_add_executor_job( + decode_media_bytes, + data, + codec, + ) async def _async_download(self, media_url: str) -> bytes: - """Download media bytes for decoding.""" - session = async_get_clientsession(self._hass) - + """Download media without buffering more than the configured limit.""" try: - async with session.get( + async with self._session.get( media_url, - timeout=ClientTimeout(total=30), + timeout=DOWNLOAD_TIMEOUT, ) as response: if not response.ok: - body = await response.text() raise HomeAssistantError( - f"Media download failed: HTTP {response.status}: " - f"{body[:200]}" + f"Media download failed: HTTP {response.status}" ) - length = response.headers.get("Content-Length") - if length and int(length) > MAX_MEDIA_BYTES: + if ( + response.content_length is not None + and response.content_length > MAX_MEDIA_BYTES + ): raise HomeAssistantError("Media file is too large") - data = await response.read() + return await _async_read_media(response) + except HomeAssistantError: + raise except (TimeoutError, ClientError, OSError) as err: raise HomeAssistantError(f"Could not download media: {media_url}") from err - if len(data) > MAX_MEDIA_BYTES: - raise HomeAssistantError("Media file is too large") - return data - - async def _async_channel(self, source: IsapiSource) -> TalkbackChannel: + async def _async_channel(self) -> TalkbackChannel: """Resolve and cache the two-way audio channel.""" if self._channel is not None: return self._channel - status, _, body = await self._async_request( - source, + body = await self._async_device_request( "GET", "/ISAPI/System/TwoWayAudio/channels", ) - if status != 200: - raise HomeAssistantError(f"Hikvision channel lookup failed: HTTP {status}") - root = ET.fromstring(body) channel_id = xml_text(root, "id") or "1" codec = xml_text(root, "audioCompressionType") @@ -187,240 +212,50 @@ async def _async_channel(self, source: IsapiSource) -> TalkbackChannel: self._channel = TalkbackChannel(channel_id, codec) return self._channel - async def _async_audio_writer( + async def _async_device_request( self, - source: IsapiSource, - path: str, - ) -> asyncio.StreamWriter: - """Open the audioData request and return its writable socket.""" - status, headers, _, writer = await self._async_raw_request( - source, - "PUT", - path, - ) - await _close_writer(writer) - - if status != 401 or "www-authenticate" not in headers: - raise HomeAssistantError( - f"Hikvision audioData challenge failed: HTTP {status}" - ) - - status, _, _, writer = await self._async_raw_request( - source, - "PUT", - path, - headers={ - "Authorization": digest_authorization( - "PUT", - path, - source.username, - source.password, - headers["www-authenticate"], - ), - "Content-Type": "application/octet-stream", - }, - keep_open=True, - read_body=False, - ) - if status != 200: - await _close_writer(writer) - raise HomeAssistantError(f"Hikvision audioData failed: HTTP {status}") - return writer - - async def _async_request( - self, - source: IsapiSource, method: str, path: str, - ) -> tuple[int, dict[str, str], bytes]: - """Send a request, retrying once with Digest auth when challenged.""" - status, headers, body, writer = await self._async_raw_request( - source, - method, - path, - ) - await _close_writer(writer) - - if status != 401 or "www-authenticate" not in headers: - return status, headers, body - - status, headers, body, writer = await self._async_raw_request( - source, - method, - path, - headers={ - "Authorization": digest_authorization( - method, - path, - source.username, - source.password, - headers["www-authenticate"], - ) - }, - ) - await _close_writer(writer) - return status, headers, body - - async def _async_raw_request( - self, - source: IsapiSource, - method: str, - path: str, - headers: dict[str, str] | None = None, - keep_open: bool = False, + *, + data: Payload | None = None, read_body: bool = True, - ) -> tuple[int, dict[str, str], bytes, asyncio.StreamWriter]: - """Send one raw HTTP request to the camera.""" + timeout: ClientTimeout = DEVICE_TIMEOUT, + ) -> bytes: + """Send an authenticated request to the Hikvision device.""" try: - reader, writer = await asyncio.wait_for( - asyncio.open_connection(source.host, source.port), - timeout=5, - ) - request_headers = { - "Host": f"{source.host}:{source.port}", - "User-Agent": "home-assistant-hikvision-player/0.1", - "Connection": "keep-alive" if keep_open else "close", - "Content-Length": "0", - } - if headers: - request_headers.update(headers) - - request = ( - f"{method} {path} HTTP/1.1\r\n" - + "".join(f"{key}: {value}\r\n" for key, value in request_headers.items()) - + "\r\n" - ) - writer.write(request.encode("ascii")) - await writer.drain() + async with self._session.request( + method, + self._base_url.with_path(path), + data=data, + timeout=timeout, + allow_redirects=False, + middlewares=(self._digest_auth,), + ) as response: + if not response.ok: + raise HomeAssistantError( + f"Hikvision ISAPI request failed: HTTP {response.status}" + ) + if read_body: + return await response.read() - status, response_headers, body = await asyncio.wait_for( - read_http_response(reader, read_body), - timeout=5, - ) - except (asyncio.TimeoutError, OSError) as err: + await response.wait_for_close() + return b"" + except HomeAssistantError: + raise + except (TimeoutError, ClientError, OSError) as err: raise HomeAssistantError( - f"Hikvision ISAPI request failed for {source.host}:{source.port}{path}" + f"Hikvision ISAPI request failed for {self._address}{path}" ) from err - return status, response_headers, body, writer - - -async def _stream_realtime( - writer: asyncio.StreamWriter, - payload: bytes, -) -> None: - """Write one-byte-per-sample G.711 audio at 8 kHz.""" - started = time.monotonic() - sent = 0 - - for offset in range(0, len(payload), CHUNK_BYTES): - chunk = payload[offset : offset + CHUNK_BYTES] - writer.write(chunk) - await writer.drain() - - sent += len(chunk) - delay = started + sent / SAMPLE_RATE - time.monotonic() - if delay > 0: - await asyncio.sleep(delay) - - -async def _close_writer(writer: asyncio.StreamWriter) -> None: - """Close a stream writer and ignore socket-close noise.""" - writer.close() - try: - await writer.wait_closed() - except OSError: - pass - - -async def read_http_response( - reader: asyncio.StreamReader, - read_body: bool, -) -> tuple[int, dict[str, str], bytes]: - """Read one simple HTTP response.""" - header = await reader.readuntil(b"\r\n\r\n") - lines = header.decode("iso-8859-1", "replace").split("\r\n") - - try: - status = int(lines[0].split()[1]) - except (IndexError, ValueError): - status = 0 - - headers: dict[str, str] = {} - for line in lines[1:]: - if ":" not in line: - continue - key, value = line.split(":", 1) - headers[key.strip().lower()] = value.strip() - - body = b"" - if read_body and "content-length" in headers: - body = await reader.readexactly(int(headers["content-length"])) - return status, headers, body - - -def digest_authorization( - method: str, - path: str, - username: str, - password: str, - challenge: str, -) -> str: - """Build a Digest Authorization header.""" - fields = digest_fields(challenge) - realm = fields.get("realm", "") - nonce = fields.get("nonce", "") - qop_raw = fields.get("qop", "") - - ha1 = md5_hex(f"{username}:{realm}:{password}") - ha2 = md5_hex(f"{method}:{path}") - parts = [ - f'Digest username="{username}"', - f'realm="{realm}"', - f'nonce="{nonce}"', - f'uri="{path}"', - ] - - if qop_raw: - qops = [qop.strip() for qop in qop_raw.split(",")] - qop = "auth" if "auth" in qops else qops[0] - nc = "00000001" - cnonce = secrets.token_hex(12) - response = md5_hex(f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}") - parts.extend( - [ - f"qop={qop}", - f"nc={nc}", - f'cnonce="{cnonce}"', - f'response="{response}"', - ] - ) - else: - parts.append(f'response="{md5_hex(f"{ha1}:{nonce}:{ha2}")}"') - - if opaque := fields.get("opaque"): - parts.append(f'opaque="{opaque}"') - - return ", ".join(parts) - - -def digest_fields(challenge: str) -> dict[str, str]: - """Parse a WWW-Authenticate Digest challenge.""" - if challenge.lower().startswith("digest"): - challenge = challenge[len("digest") :].strip() - fields: dict[str, str] = {} - for match in re.finditer(r'(\w+)=("(?:[^"\\]|\\.)*"|[^,]+)', challenge): - value = match.group(2).strip() - if len(value) >= 2 and value[0] == '"' and value[-1] == '"': - value = value[1:-1] - fields[match.group(1)] = value - return fields - - -def md5_hex(value: str) -> str: - """Return MD5 hex for HTTP Digest auth.""" - return hashlib.md5(value.encode("utf-8")).hexdigest() +async def _async_read_media(response: ClientResponse) -> bytes: + """Read a media response while enforcing the download limit.""" + data = bytearray() + async for chunk in response.content.iter_chunked(DOWNLOAD_CHUNK_BYTES): + if len(data) + len(chunk) > MAX_MEDIA_BYTES: + raise HomeAssistantError("Media file is too large") + data.extend(chunk) + return bytes(data) def xml_text(root: ET.Element, tag: str) -> str: diff --git a/custom_components/hikvision_player/manifest.json b/custom_components/hikvision_player/manifest.json index 2ef1081..00fb412 100644 --- a/custom_components/hikvision_player/manifest.json +++ b/custom_components/hikvision_player/manifest.json @@ -13,5 +13,5 @@ "requirements": [ "miniaudio==1.71" ], - "version": "0.1.2" + "version": "0.1.3" } diff --git a/custom_components/hikvision_player/media_player.py b/custom_components/hikvision_player/media_player.py index 3a80ad5..1749b9d 100644 --- a/custom_components/hikvision_player/media_player.py +++ b/custom_components/hikvision_player/media_player.py @@ -2,9 +2,6 @@ from __future__ import annotations -from pathlib import Path -from urllib.parse import unquote - import voluptuous as vol from homeassistant.components import media_source @@ -22,6 +19,7 @@ CONF_PORT, CONF_USERNAME, STATE_IDLE, + STATE_PLAYING, ) from homeassistant.core import HomeAssistant from homeassistant.helpers import config_validation as cv @@ -128,11 +126,18 @@ async def async_play_media( ) media_id = sourced_media.url - media_url = self._local_media_path(media_id) or async_process_play_media_url( + media_url = async_process_play_media_url( self.hass, media_id, ) - await self._client.async_play(media_url) + self._attr_state = STATE_PLAYING + self.async_write_ha_state() + + try: + await self._client.async_play(media_url) + finally: + self._attr_state = STATE_IDLE + self.async_write_ha_state() async def async_media_stop(self) -> None: """Stop current playback.""" @@ -150,18 +155,8 @@ async def async_browse_media( media_content_id: str | None = None, ) -> BrowseMedia: """Browse Home Assistant media sources.""" - return await media_source.async_browse_media(self.hass, media_content_id) - - def _local_media_path(self, media_id: str) -> str | None: - """Resolve Home Assistant /local media to a local file path.""" - if not media_id.startswith("/local/"): - return None - - www_dir = Path(self.hass.config.path("www")).resolve() - relative_path = unquote(media_id.removeprefix("/local/")).lstrip("/") - candidate = (www_dir / relative_path).resolve() - - if not candidate.is_relative_to(www_dir) or not candidate.is_file(): - return None - - return str(candidate) + return await media_source.async_browse_media( + self.hass, + media_content_id, + content_filter=lambda item: item.media_content_type.startswith("audio/"), + )