get_folder_size,
get_free_space,
get_free_space_percentage,
- get_ip,
- get_ips,
+ get_ip_addresses,
select_free_port,
try_parse_bool,
)
values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
"""Return all Config Entries for this core module (if any)."""
- default_ip = await get_ip()
- all_ips = await get_ips()
+ ip_addresses = await get_ip_addresses()
default_port = await select_free_port(8097, 9200)
return (
+ ConfigEntry(
+ key=CONF_PUBLISH_IP,
+ type=ConfigEntryType.STRING,
+ default_value=ip_addresses[0],
+ label="Published IP address",
+ description="This IP address is communicated to players where to find this server."
+ "\nMake sure that this IP can be reached by players on the local network, "
+ "otherwise audio streaming will not work.",
+ required=False,
+ ),
ConfigEntry(
key=CONF_BIND_PORT,
type=ConfigEntryType.INTEGER,
label="Fixed/fallback gain adjustment for tracks",
category="audio",
),
- ConfigEntry(
- key=CONF_PUBLISH_IP,
- type=ConfigEntryType.STRING,
- default_value=default_ip,
- label="Published IP address",
- description="This IP address is communicated to players where to find this server. "
- "Override the default in advanced scenarios, such as multi NIC configurations. \n"
- "Make sure that this server can be reached "
- "on the given IP and TCP port by players on the local network. \n"
- "This is an advanced setting that should normally "
- "not be adjusted in regular setups.",
- category="advanced",
- required=False,
- ),
ConfigEntry(
key=CONF_BIND_IP,
type=ConfigEntryType.STRING,
default_value="0.0.0.0",
- options=[ConfigValueOption(x, x) for x in {"0.0.0.0", *all_ips}],
+ options=[ConfigValueOption(x, x) for x in {"0.0.0.0", *ip_addresses}],
label="Bind to IP/interface",
description="Start the stream server on this specific interface. \n"
"Use 0.0.0.0 to bind to all interfaces, which is the default. \n"
from music_assistant.helpers.api import APICommandHandler, parse_arguments
from music_assistant.helpers.audio import get_preview_stream
from music_assistant.helpers.json import json_dumps
-from music_assistant.helpers.util import get_ip, get_ips
+from music_assistant.helpers.util import get_ip_addresses
from music_assistant.helpers.webserver import Webserver
from music_assistant.models.core_controller import CoreController
values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
"""Return all Config Entries for this core module (if any)."""
- default_publish_ip = await get_ip()
+ ip_addresses = await get_ip_addresses()
+ default_publish_ip = ip_addresses[0]
if self.mass.running_as_hass_addon:
return (
ConfigEntry(
# HA supervisor not present: user is responsible for securing the webserver
# we give the tools to do so by presenting config options
- all_ips = await get_ips()
default_base_url = f"http://{default_publish_ip}:{DEFAULT_SERVER_PORT}"
return (
ConfigEntry(
key=CONF_BIND_IP,
type=ConfigEntryType.STRING,
default_value="0.0.0.0",
- options=[ConfigValueOption(x, x) for x in {"0.0.0.0", *all_ips}],
+ options=[ConfigValueOption(x, x) for x in {"0.0.0.0", *ip_addresses}],
label="Bind to IP/interface",
description="Start the (web)server on this specific interface. \n"
"Use 0.0.0.0 to bind to all interfaces. \n"
# add jsonrpc api
routes.append(("POST", "/api", self._handle_jsonrpc_api_command))
# start the webserver
- default_publish_ip = await get_ip()
+ all_ip_addresses = await get_ip_addresses()
+ default_publish_ip = all_ip_addresses[0]
if self.mass.running_as_hass_addon:
# if we're running on the HA supervisor the webserver is secured by HA ingress
# we only start the webserver on the internal docker network and ingress connects
else:
# use internal ("172.30.32.) IP
self.publish_ip = bind_ip = next(
- (x for x in await get_ips() if x.startswith("172.30.32.")), default_publish_ip
+ (x for x in all_ip_addresses if x.startswith("172.30.32.")), default_publish_ip
)
base_url = f"http://{self.publish_ip}:{self.publish_port}"
else:
return line
-async def get_ip() -> str:
- """Get primary IP-address for this host."""
+async def get_ip_addresses(include_ipv6: bool = False) -> tuple[str]:
+ """Return all IP-adresses of all network interfaces."""
- def _get_ip() -> str:
- """Get primary IP-address for this host."""
- sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
- try:
- # doesn't even have to be reachable
- sock.connect(("10.255.255.255", 1))
- _ip = str(sock.getsockname()[0])
- except Exception:
- _ip = "127.0.0.1"
- finally:
- sock.close()
- return _ip
+ def call() -> set[str]:
+ result: list[tuple[int, str]] = []
+ adapters = ifaddr.get_adapters()
+ for adapter in adapters:
+ for ip in adapter.ips:
+ if ip.is_IPv6 and not include_ipv6:
+ continue
+ if ip.ip.startswith(("127", "169.254")):
+ # filter out IPv4 loopback/APIPA address
+ continue
+ if ip.ip.startswith(("::1", "::ffff:", "fe80")):
+ # filter out IPv6 loopback/link-local address
+ continue
+ if ip.ip.startswith(("192.", "10.")):
+ score = 2
+ elif ip.ip.startswith("172."):
+ # we rank the 172 range a bit lower as its most
+ # often used as the private docker network
+ score = 1
+ else:
+ score = 0
+ result.append((score, ip.ip))
+ result.sort(key=lambda x: x[0], reverse=True)
+ return tuple(ip[1] for ip in result)
- return await asyncio.to_thread(_get_ip)
+ return await asyncio.to_thread(call)
async def is_port_in_use(port: int) -> bool:
return await asyncio.to_thread(_resolve)
-async def get_ip_pton(ip_string: str | None = None) -> bytes:
- """Return socket pton for local ip."""
- if ip_string is None:
- ip_string = await get_ip()
+async def get_ip_pton(ip_string: str) -> bytes:
+ """Return socket pton for a local ip."""
try:
return await asyncio.to_thread(socket.inet_pton, socket.AF_INET, ip_string)
except OSError:
return None
-async def get_ips(include_ipv6: bool = False, ignore_loopback: bool = True) -> set[str]:
- """Return all IP-adresses of all network interfaces."""
-
- def call() -> set[str]:
- result: set[str] = set()
- adapters = ifaddr.get_adapters()
- for adapter in adapters:
- for ip in adapter.ips:
- if ip.is_IPv6 and not include_ipv6:
- continue
- if ip.ip == "127.0.0.1" and ignore_loopback:
- continue
- result.add(ip.ip)
- return result
-
- return await asyncio.to_thread(call)
-
-
async def is_hass_supervisor() -> bool:
"""Return if we're running inside the HA Supervisor (e.g. HAOS)."""
projects / music-assistant-server.git / commitdiff