from music_assistant.helpers.ffmpeg import check_ffmpeg_version, get_ffmpeg_stream
from music_assistant.helpers.util import (
divide_chunks,
+ format_ip_for_url,
get_ip_addresses,
get_total_system_memory,
select_free_port,
values: dict[str, ConfigValueType] | None = None,
) -> tuple[ConfigEntry, ...]:
"""Return all Config Entries for this core module (if any)."""
- ip_addresses = await get_ip_addresses()
+ ip_addresses = await get_ip_addresses(include_ipv6=True)
default_port = await select_free_port(8097, 9200)
return (
ConfigEntry(
await self._server.setup(
bind_ip=bind_ip,
bind_port=cast("int", self.publish_port),
- base_url=f"http://{self.publish_ip}:{self.publish_port}",
+ base_url=f"http://{format_ip_for_url(str(self.publish_ip))}:{self.publish_port}",
static_routes=[
(
"*",
from music_assistant.helpers.audio import get_preview_stream
from music_assistant.helpers.json import json_dumps, json_loads
from music_assistant.helpers.redirect_validation import is_allowed_redirect_url
-from music_assistant.helpers.util import get_ip_addresses
+from music_assistant.helpers.util import format_ip_for_url, 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)."""
- ip_addresses = await get_ip_addresses()
+ ip_addresses = await get_ip_addresses(include_ipv6=True)
default_publish_ip = ip_addresses[0]
# Handle verify SSL action
# Determine if SSL is enabled from values
ssl_enabled = values.get(CONF_ENABLE_SSL, False) if values else False
protocol = "https" if ssl_enabled else "http"
- default_base_url = f"{protocol}://{default_publish_ip}:{DEFAULT_SERVER_PORT}"
+ default_base_url = (
+ f"{protocol}://{format_ip_for_url(default_publish_ip)}:{DEFAULT_SERVER_PORT}"
+ )
return (
ConfigEntry(
key=CONF_AUTH_ALLOW_SELF_REGISTRATION,
routes.append(("GET", "/sendspin", self._sendspin_proxy.handle_sendspin_proxy))
await self.auth.setup()
# start the webserver
- all_ip_addresses = await get_ip_addresses()
+ all_ip_addresses = await get_ip_addresses(include_ipv6=True)
default_publish_ip = all_ip_addresses[0]
if self.mass.running_as_hass_addon:
# if we're running on the HA supervisor we start an additional TCP site
"""Check if port is in use."""
def _is_port_in_use() -> bool:
- with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as _sock:
- # Set SO_REUSEADDR to match asyncio.start_server behavior
- # This allows binding to ports in TIME_WAIT state
- _sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ # Try both IPv4 and IPv6 to support single-stack and dual-stack systems.
+ # A port is considered free if it can be bound on at least one address family.
+ for family, addr in ((socket.AF_INET, "0.0.0.0"), (socket.AF_INET6, "::")):
try:
- _sock.bind(("0.0.0.0", port))
+ with socket.socket(family, socket.SOCK_STREAM) as _sock:
+ # Set SO_REUSEADDR to match asyncio.start_server behavior
+ # This allows binding to ports in TIME_WAIT state
+ _sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ _sock.bind((addr, port))
+ return False
except OSError:
- return True
- return False
+ continue
+ return True
return await asyncio.to_thread(_is_port_in_use)
return await asyncio.to_thread(socket.inet_pton, socket.AF_INET6, ip_string)
+def format_ip_for_url(ip_address: str) -> str:
+ """Wrap IPv6 addresses in brackets for use in URLs (RFC 2732)."""
+ if ":" in ip_address:
+ return f"[{ip_address}]"
+ return ip_address
+
+
async def get_folder_size(folderpath: str) -> float:
"""Return folder size in gb."""
# filter out APIPA address
continue
return address
+ # fall back to IPv6 addresses if no usable IPv4 address found
+ for address in discovery_info.parsed_addresses(IPVersion.V6Only):
+ if address.startswith(("::1", "fe80")):
+ # filter out loopback and link-local addresses
+ continue
+ return address
return None
self.config = ConfigController(self)
await self.config.setup()
# create shared zeroconf instance
- # TODO: enumerate interfaces and enable IPv6 support
zeroconf_interfaces = self.config.get_raw_core_config_value(
"streams", CONF_ZEROCONF_INTERFACES, "default"
)
self.aiozc = AsyncZeroconf(
- ip_version=IPVersion.V4Only,
+ ip_version=IPVersion.All,
interfaces=InterfaceChoice.All
if zeroconf_interfaces == "all"
else InterfaceChoice.Default,
# for AirPlay 2 (HAP) pair-verify to work with previously paired devices
self.dacp_id = dacp_id = self.mass.server_id[:16].upper()
self.logger.debug("Starting DACP ActiveRemote %s on port %s", dacp_id, dacp_port)
- self._dacp_server = await asyncio.start_server(
- self._handle_dacp_request, "0.0.0.0", dacp_port
- )
+ self._dacp_server = await asyncio.start_server(self._handle_dacp_request, port=dacp_port)
server_id = f"iTunes_Ctrl_{dacp_id}.{DACP_DISCOVERY_TYPE}"
self._dacp_info = AsyncServiceInfo(
DACP_DISCOVERY_TYPE,
projects / music-assistant-server.git / commitdiff