/
/
/
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