Guard linking credential that is already linked (#57595)

* Guard linking credential that is already linked

* Update test descriptions

homeassistant/auth/__init__.py

@@ -276,6 +276,12 @@ class AuthManager:
         self, user: models.User, credentials: models.Credentials
     ) -> None:
         """Link credentials to an existing user."""
+        linked_user = await self.async_get_user_by_credentials(credentials)
+        if linked_user == user:
+            return
+        if linked_user is not None:
+            raise ValueError("Credential is already linked to a user")
+
         await self._store.async_link_user(user, credentials)
 
     async def async_remove_user(self, user: models.User) -> None:

homeassistant/components/auth/__init__.py

@@ -412,6 +412,14 @@ class LinkUserView(HomeAssistantView):
         if credentials is None:
             return self.json_message("Invalid code", status_code=HTTPStatus.BAD_REQUEST)
 
+        linked_user = await hass.auth.async_get_user_by_credentials(credentials)
+        if linked_user != user and linked_user is not None:
+            return self.json_message(
+                "Credential already linked", status_code=HTTPStatus.BAD_REQUEST
+            )
+
+        # No-op if credential is already linked to the user it will be linked to
+        if linked_user != user:
             await hass.auth.async_link_user(user, credentials)
         return self.json_message("User linked")
 

tests/auth/test_init.py

@@ -288,6 +288,16 @@ async def test_linking_user_to_two_auth_providers(hass, hass_storage):
     await manager.async_link_user(user, new_credential)
     assert len(user.credentials) == 2
 
+    # Linking it again to same user is a no-op
+    await manager.async_link_user(user, new_credential)
+    assert len(user.credentials) == 2
+
+    # Linking a credential to a user while the credential is already linked to another user should raise
+    user_2 = await manager.async_create_user("User 2")
+    with pytest.raises(ValueError):
+        await manager.async_link_user(user_2, new_credential)
+    assert len(user_2.credentials) == 0
+
 
 async def test_saving_loading(hass, hass_storage):
     """Test storing and saving data.

tests/components/auth/test_init_link_user.py

@@ -1,4 +1,6 @@
 """Tests for the link user flow."""
+from unittest.mock import patch
+
 from . import async_setup_auth
 
 from tests.common import CLIENT_ID, CLIENT_REDIRECT_URI
@@ -122,3 +124,48 @@ async def test_link_user_invalid_auth(hass, aiohttp_client):
 
     assert resp.status == 401
     assert len(info["user"].credentials) == 0
+
+
+async def test_link_user_already_linked_same_user(hass, aiohttp_client):
+    """Test linking a user to a credential it's already linked to."""
+    info = await async_get_code(hass, aiohttp_client)
+    client = info["client"]
+    code = info["code"]
+
+    # Link user
+    with patch.object(
+        hass.auth, "async_get_user_by_credentials", return_value=info["user"]
+    ):
+        resp = await client.post(
+            "/auth/link_user",
+            json={"client_id": CLIENT_ID, "code": code},
+            headers={"authorization": f"Bearer {info['access_token']}"},
+        )
+
+    assert resp.status == 200
+    # The credential was not added because it saw that it was already linked
+    assert len(info["user"].credentials) == 0
+
+
+async def test_link_user_already_linked_other_user(hass, aiohttp_client):
+    """Test linking a user to a credential already linked to other user."""
+    info = await async_get_code(hass, aiohttp_client)
+    client = info["client"]
+    code = info["code"]
+
+    another_user = await hass.auth.async_create_user(name="Another")
+
+    # Link user
+    with patch.object(
+        hass.auth, "async_get_user_by_credentials", return_value=another_user
+    ):
+        resp = await client.post(
+            "/auth/link_user",
+            json={"client_id": CLIENT_ID, "code": code},
+            headers={"authorization": f"Bearer {info['access_token']}"},
+        )
+
+    assert resp.status == 400
+    # The credential was not added because it saw that it was already linked
+    assert len(info["user"].credentials) == 0
+    assert len(another_user.credentials) == 0