diff --git a/authentik/stages/user_write/stage.py b/authentik/stages/user_write/stage.py
index 2319e7414..4de0f8135 100644
--- a/authentik/stages/user_write/stage.py
+++ b/authentik/stages/user_write/stage.py
@@ -124,7 +124,8 @@ class UserWriteStageView(StageView):
             connection: UserSourceConnection = self.executor.plan.context[
                 PLAN_CONTEXT_SOURCES_CONNECTION
             ]
-            user.attributes[USER_ATTRIBUTE_SOURCES].append(connection.source.name)
+            if connection.source.name not in user.attributes[USER_ATTRIBUTE_SOURCES]:
+                user.attributes[USER_ATTRIBUTE_SOURCES].append(connection.source.name)
 
     def get(self, request: HttpRequest) -> HttpResponse:
         """Save data in the current flow to the currently pending user. If no user is pending,
diff --git a/authentik/stages/user_write/tests.py b/authentik/stages/user_write/tests.py
index 9bbfcbcbf..66084f67e 100644
--- a/authentik/stages/user_write/tests.py
+++ b/authentik/stages/user_write/tests.py
@@ -97,6 +97,47 @@ class TestUserWriteStage(FlowTestCase):
         self.assertEqual(user_qs.first().attributes["foo"], "bar")
         self.assertNotIn("some_ignored_attribute", user_qs.first().attributes)
 
+    def test_user_update_source(self):
+        """Test update of existing user with a source"""
+        new_password = generate_key()
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        plan.context[PLAN_CONTEXT_PENDING_USER] = User.objects.create(
+            username="unittest",
+            email="test@goauthentik.io",
+            attributes={
+                USER_ATTRIBUTE_SOURCES: [
+                    self.source.name,
+                ]
+            },
+        )
+        plan.context[PLAN_CONTEXT_SOURCES_CONNECTION] = UserSourceConnection(source=self.source)
+        plan.context[PLAN_CONTEXT_PROMPT] = {
+            "username": "test-user-new",
+            "password": new_password,
+            "attributes.some.custom-attribute": "test",
+            "attributes": {
+                "foo": "bar",
+            },
+            "some_ignored_attribute": "bar",
+        }
+        session = self.client.session
+        session[SESSION_KEY_PLAN] = plan
+        session.save()
+
+        response = self.client.post(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug})
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertStageRedirects(response, reverse("authentik_core:root-redirect"))
+        user_qs = User.objects.filter(username=plan.context[PLAN_CONTEXT_PROMPT]["username"])
+        self.assertTrue(user_qs.exists())
+        self.assertTrue(user_qs.first().check_password(new_password))
+        self.assertEqual(user_qs.first().attributes["some"]["custom-attribute"], "test")
+        self.assertEqual(user_qs.first().attributes["foo"], "bar")
+        self.assertEqual(user_qs.first().attributes[USER_ATTRIBUTE_SOURCES], [self.source.name])
+        self.assertNotIn("some_ignored_attribute", user_qs.first().attributes)
+
     @patch(
         "authentik.flows.views.executor.to_stage_response",
         TO_STAGE_RESPONSE_MOCK,