continue
if re.match(r"^[•\-\*]\s*Chore\(deps", stripped_line, re.IGNORECASE):
continue
+ # Skip "No changes" entries
+ if re.match(r"^[•\-\*]\s*No changes\s*$", stripped_line, re.IGNORECASE):
+ continue
# Add the change
frontend_changes.append(stripped_line)
"""
Link a user to an authentication provider.
+ If a link already exists for this provider/provider_user_id, returns the existing link.
+
:param user: The user to link.
:param provider_type: The provider type.
:param provider_user_id: The user ID from the provider (e.g., password hash, OAuth ID).
"""
+ # Check if a link already exists for this provider/provider_user_id
+ existing_link = await self.database.get_row(
+ "user_auth_providers",
+ {
+ "provider_type": provider_type.value,
+ "provider_user_id": provider_user_id,
+ },
+ )
+
+ if existing_link:
+ # Link already exists - return it
+ return UserAuthProvider(
+ link_id=existing_link["link_id"],
+ user_id=existing_link["user_id"],
+ provider_type=AuthProviderType(existing_link["provider_type"]),
+ provider_user_id=existing_link["provider_user_id"],
+ created_at=datetime.fromisoformat(existing_link["created_at"]),
+ )
+
+ # Create new link
link_id = secrets.token_urlsafe(32)
created_at = utc()
link_data = {
if not self.mass.config.onboard_done:
self.logger.warning(
"\n\n################################################################################\n"
- "Starting webserver on %s:%s - base url: %s\n"
- "If this is incorrect, see the documentation how to configure the Webserver\n"
- "in Settings --> Core modules --> Webserver\n"
+ "### SETUP REQUIRED ###\n"
+ "################################################################################\n"
+ "\n"
+ "Music Assistant is running in setup mode.\n"
+ "Please complete the setup by visiting:\n"
+ "\n"
+ " %s/setup\n"
+ "\n"
+ "Webserver running on: %s:%s\n"
+ "\n"
+ "If this address is incorrect, see the documentation on how to configure\n"
+ "the Webserver in Settings --> Core modules --> Webserver\n"
+ "\n"
"################################################################################\n",
+ base_url,
bind_ip,
self.publish_port,
- base_url,
)
else:
self.logger.info(
async def _handle_jsonrpc_api_command(self, request: web.Request) -> web.Response:
"""Handle incoming JSON RPC API command."""
+ # Block until onboarding is complete
+ if not self.mass.config.onboard_done:
+ return web.Response(status=503, text="Setup required")
+
if not request.can_read_body:
return web.Response(status=400, text="Body required")
cmd_data = await request.read()
async def _handle_auth_login(self, request: web.Request) -> web.Response:
"""Handle login request."""
+ # Block until onboarding is complete
+ if not self.mass.config.onboard_done:
+ return web.json_response(
+ {"success": False, "error": "Setup required"},
+ status=403,
+ headers={
+ "Access-Control-Allow-Origin": "*",
+ "Access-Control-Allow-Methods": "POST, OPTIONS",
+ "Access-Control-Allow-Headers": "Content-Type, Authorization",
+ },
+ )
+
try:
if not request.can_read_body:
return web.Response(status=400, text="Body required")
)
if not user:
- # Security: Ensure at least one user exists (setup should have been completed)
- if not await mass.webserver.auth.has_users():
- # No users exist - setup has not been completed
- # This should not happen as the server redirects to /setup
+ # Only auto-create users after onboarding is complete
+ if not mass.config.onboard_done:
return None
- # Auto-create user for Ingress (they're already authenticated by HA)
- # Always create with USER role (admin is created during setup)
- user = await mass.webserver.auth.create_user(
- username=ingress_username,
- role=UserRole.USER,
- display_name=ingress_display_name,
- )
- # Link to Home Assistant provider
+ # Check if a user with this username already exists
+ user = await mass.webserver.auth.get_user_by_username(ingress_username)
+
+ if not user:
+ # Auto-create user for Ingress (they're already authenticated by HA)
+ # Always create with USER role (admin is created during setup)
+ user = await mass.webserver.auth.create_user(
+ username=ingress_username,
+ role=UserRole.USER,
+ display_name=ingress_display_name,
+ )
+
+ # Link to Home Assistant provider (or create the link if user already existed)
await mass.webserver.auth.link_user_to_provider(
user, AuthProviderType.HOME_ASSISTANT, ingress_user_id
)
from __future__ import annotations
+from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING, cast
self._remote_id: str | None = None
self._enabled: bool = False
self._using_ha_cloud: bool = False
+ self._on_unload_callbacks: list[Callable[[], None]] = []
async def setup(self) -> None:
"""Initialize the remote access manager."""
async def close(self) -> None:
"""Cleanup on exit."""
await self.stop()
+ for unload_cb in self._on_unload_callbacks:
+ unload_cb()
async def start(self) -> None:
"""Start the remote access gateway."""
await self.stop()
return await get_remote_access_info()
- self.mass.register_api_command("remote_access/info", get_remote_access_info)
- self.mass.register_api_command(
- "remote_access/configure", configure_remote_access, required_role="admin"
+ self._on_unload_callbacks.append(
+ self.mass.register_api_command("remote_access/info", get_remote_access_info)
+ )
+ self._on_unload_callbacks.append(
+ self.mass.register_api_command(
+ "remote_access/configure", configure_remote_access, required_role="admin"
+ )
)
server_info = self.mass.get_server_info()
await self._send_message(server_info)
+ # Block until onboarding is complete
+ if not self.mass.config.onboard_done:
+ await self._send_message(ErrorResultMessage("connection", 503, "Setup required"))
+ await wsock.close()
+ return wsock
+
# For Ingress connections, auto-create/link user and subscribe to events immediately
# For regular connections, events will be subscribed after successful authentication
if self._is_ingress:
)
if not user:
- # Security: Ensure at least one user exists (setup should have been completed)
- if not await self.webserver.auth.has_users():
- # No users exist - setup has not been completed
- # This should not happen as the server redirects to /setup
- self._logger.warning("Ingress connection attempted before setup completed")
+ # Only auto-create users after onboarding is complete
+ if not self.mass.config.onboard_done:
+ self._logger.warning("Ingress connection attempted before setup")
return
- # Auto-create user for Ingress (they're already authenticated by HA)
- # Always create with USER role (admin is created during setup)
- user = await self.webserver.auth.create_user(
- username=ingress_username,
- role=UserRole.USER,
- display_name=ingress_display_name,
- )
- # Link to Home Assistant provider
+ # Check if a user with this username already exists
+ user = await self.webserver.auth.get_user_by_username(ingress_username)
+
+ if not user:
+ # Auto-create user for Ingress (they're already authenticated by HA)
+ # Always create with USER role (admin is created during setup)
+ user = await self.webserver.auth.create_user(
+ username=ingress_username,
+ role=UserRole.USER,
+ display_name=ingress_display_name,
+ )
+
+ # Link to Home Assistant provider (or create the link if user already existed)
await self.webserver.auth.link_user_to_provider(
user, AuthProviderType.HOME_ASSISTANT, ingress_user_id
)
# Username should be normalized to lowercase
assert updated_user is not None
assert updated_user.username == "updateduser"
+
+
+async def test_link_user_to_provider_idempotent(auth_manager: AuthenticationManager) -> None:
+ """Test that linking user to provider is idempotent.
+
+ This tests the fix for the bug where re-linking a user would cause
+ IntegrityError due to UNIQUE constraint on (provider_type, provider_user_id).
+
+ :param auth_manager: AuthenticationManager instance.
+ """
+ user = await auth_manager.create_user(username="hauser", role=UserRole.USER)
+
+ # Link user to Home Assistant provider for the first time
+ link1 = await auth_manager.link_user_to_provider(
+ user,
+ AuthProviderType.HOME_ASSISTANT,
+ "ha_user_456",
+ )
+
+ assert link1 is not None
+ assert link1.user_id == user.user_id
+ assert link1.provider_type == AuthProviderType.HOME_ASSISTANT
+ assert link1.provider_user_id == "ha_user_456"
+
+ # Linking the same user again should return existing link without error
+ link2 = await auth_manager.link_user_to_provider(
+ user,
+ AuthProviderType.HOME_ASSISTANT,
+ "ha_user_456",
+ )
+
+ assert link2 is not None
+ assert link2.link_id == link1.link_id # Should be same link
+ assert link2.user_id == user.user_id
+ assert link2.provider_type == AuthProviderType.HOME_ASSISTANT
+ assert link2.provider_user_id == "ha_user_456"
+
+
+async def test_ingress_auth_existing_username(auth_manager: AuthenticationManager) -> None:
+ """Test HA ingress auth when username exists but isn't linked to HA provider.
+
+ This tests the scenario where a user is created during setup, and then
+ tries to login via HA ingress with the same username.
+
+ :param auth_manager: AuthenticationManager instance.
+ """
+ # Simulate user created during initial setup
+ existing_user = await auth_manager.create_user(
+ username="admin",
+ role=UserRole.ADMIN,
+ display_name="Admin User",
+ )
+
+ # Now simulate HA ingress trying to auto-create a user with same username
+ # This should find the existing user and link it instead of creating new one
+ user = await auth_manager.get_user_by_username("admin")
+ assert user is not None
+ assert user.user_id == existing_user.user_id
+
+ # Link the existing user to HA provider (what ingress flow would do)
+ link = await auth_manager.link_user_to_provider(
+ user,
+ AuthProviderType.HOME_ASSISTANT,
+ "ha_admin_123",
+ )
+
+ assert link is not None
+ assert link.user_id == existing_user.user_id
+
+ # Verify we can retrieve user by provider link
+ retrieved_user = await auth_manager.get_user_by_provider_link(
+ AuthProviderType.HOME_ASSISTANT,
+ "ha_admin_123",
+ )
+
+ assert retrieved_user is not None
+ assert retrieved_user.user_id == existing_user.user_id
+ assert retrieved_user.username == "admin"
projects / music-assistant-server.git / commitdiff