music-assistant-server

1"""FullyKiosk Player provider for Music Assistant."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import re
8from dataclasses import dataclass
9from typing import Any
10
11from aiohttp import ClientSession, Fingerprint
12from fullykiosk import FullyKiosk
13from music_assistant_models.errors import SetupFailedError
14
15from music_assistant.constants import (
16    CONF_IP_ADDRESS,
17    CONF_PASSWORD,
18    CONF_PORT,
19    CONF_SSL_FINGERPRINT,
20    CONF_USE_SSL,
21    CONF_VERIFY_SSL,
22    VERBOSE_LOG_LEVEL,
23)
24from music_assistant.models.player_provider import PlayerProvider
25
26from .player import FullyKioskPlayer
27
28
29@dataclass
30class _FingerprintSessionWrapper:
31    """Proxy ClientSession that enforces a TLS fingerprint."""
32
33    session: ClientSession
34    fingerprint: Fingerprint
35
36    def get(self, *args: Any, **kwargs: Any) -> Any:
37        """Call the wrapped session.get while injecting the fingerprint."""
38        kwargs.setdefault("ssl", self.fingerprint)
39        return self.session.get(*args, **kwargs)
40
41    def __getattr__(self, name: str) -> Any:
42        """Delegate attribute access to the wrapped session."""
43        return getattr(self.session, name)
44
45
46def _build_fingerprint(value: str) -> Fingerprint:
47    """Parse a fingerprint string (sha256 hex) into an aiohttp Fingerprint."""
48    normalized = re.sub(r"[^0-9a-fA-F]", "", value).lower()
49    if not normalized:
50        msg = "Empty fingerprint provided."
51        raise ValueError(msg)
52    if len(normalized) % 2 != 0:
53        msg = "Fingerprint must contain an even number of hex characters."
54        raise ValueError(msg)
55    digest = bytes.fromhex(normalized)
56    return Fingerprint(digest)
57
58
59class FullyKioskProvider(PlayerProvider):
60    """Player provider for FullyKiosk based players."""
61
62    async def handle_async_init(self) -> None:
63        """Handle async initialization of the provider."""
64        # set-up fullykiosk logging
65        if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
66            logging.getLogger("fullykiosk").setLevel(logging.DEBUG)
67        else:
68            logging.getLogger("fullykiosk").setLevel(self.logger.level + 10)
69
70        use_ssl = bool(self.config.get_value(CONF_USE_SSL))
71        fingerprint_value = self.config.get_value(CONF_SSL_FINGERPRINT)
72        fingerprint_raw = fingerprint_value.strip() if isinstance(fingerprint_value, str) else ""
73        if fingerprint_raw and not use_ssl:
74            msg = "Fingerprint validation requires HTTPS to be enabled."
75            raise SetupFailedError(msg)
76
77        verify_ssl = bool(self.config.get_value(CONF_VERIFY_SSL)) if use_ssl else False
78        http_session: ClientSession | _FingerprintSessionWrapper
79        if use_ssl:
80            if fingerprint_raw:
81                try:
82                    fingerprint = _build_fingerprint(fingerprint_raw)
83                except ValueError as err:
84                    msg = f"Invalid TLS fingerprint configured: {err}"
85                    raise SetupFailedError(msg) from err
86                http_session = _FingerprintSessionWrapper(self.mass.http_session, fingerprint)
87                verify_ssl = True
88            else:
89                http_session = (
90                    self.mass.http_session if verify_ssl else self.mass.http_session_no_ssl
91                )
92        else:
93            http_session = self.mass.http_session_no_ssl
94
95        fully_kiosk = FullyKiosk(
96            http_session,
97            self.config.get_value(CONF_IP_ADDRESS),
98            self.config.get_value(CONF_PORT),
99            self.config.get_value(CONF_PASSWORD),
100            use_ssl=use_ssl,
101            verify_ssl=verify_ssl,
102        )
103        try:
104            async with asyncio.timeout(15):
105                await fully_kiosk.getDeviceInfo()
106        except Exception as err:
107            msg = f"Unable to start the FullyKiosk connection ({err!s}"
108            raise SetupFailedError(msg) from err
109        player_id = fully_kiosk.deviceInfo["deviceID"]
110        scheme = "https" if use_ssl else "http"
111        address = (
112            f"{scheme}://{self.config.get_value(CONF_IP_ADDRESS)}:"
113            f"{self.config.get_value(CONF_PORT)}"
114        )
115        player = FullyKioskPlayer(self, player_id, fully_kiosk, address)
116        player.set_attributes()
117        await self.mass.players.register(player)
118