sources: allow the use of a custom identifier

[rissson]

Details

Stuff to figure out:

• how do we do this with LDAP, if at all?
• Plex doesn't seem to use a SourceFlowManager for authenticated linking
• how do we do this for SCIM

Closes #13773

---

Checklist

• Local tests pass (ak test authentik/)
• The code has been formatted (make lint-fix)

If an API change has been made

• The API schema has been updated (make gen-build)

If changes to the frontend have been made

• The code has been formatted (make web)

If applicable

• The documentation has been updated
• The documentation has been formatted (make docs)

[netlify]

✅ Deploy Preview for authentik-storybook canceled.

Name	Link
🔨 Latest commit	e214eba
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-storybook/deploys/68823bdcd655de00088a875f

[netlify]

✅ Deploy Preview for authentik-integrations canceled.

Name	Link
🔨 Latest commit	e214eba
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-integrations/deploys/68823bddbb64570008559209

[netlify]

✅ Deploy Preview for authentik-docs canceled.

Name	Link
🔨 Latest commit	e214eba
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-docs/deploys/68823bdd8c6fe400086478d4

[codecov]

❌ 11 Tests Failed:

Tests completed	Failed	Passed	Skipped
2025	11	2014	2

View the top 3 failed test(s) by shortest run time

authentik.core.tests.test_source_flow_manager.TestSourceFlowManager::test_unauthenticated_enroll

Stack Traces | 0.029s run time

self = <unittest.case._Outcome object at 0x7fa112b2a510>
test_case = <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_enroll>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.13.5........./x64/lib/python3.13/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_enroll>
result = <TestCaseFunction test_unauthenticated_enroll>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
                    self._callSetUp()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.13.5........./x64/lib/python3.13/unittest/case.py:651: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_enroll>
method = <bound method TestSourceFlowManager.test_unauthenticated_enroll of <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_enroll>>

    def _callTestMethod(self, method):
>       if method() is not None:

.../hostedtoolcache/Python/3.13.5........./x64/lib/python3.13/unittest/case.py:606: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_enroll>

    def test_unauthenticated_enroll(self):
        """Test un-authenticated user enrolling"""
        request = get_request("/", user=AnonymousUser())
>       flow_manager = OAuthSourceFlowManager(
            self.source, request, self.identifier, {"info": {}}, {}
        )
E       TypeError: SourceFlowManager.__init__() takes 5 positional arguments but 6 were given

.../core/tests/test_source_flow_manager.py:41: TypeError

authentik.core.tests.test_source_flow_manager.TestSourceFlowManager::test_unauthenticated_enroll_link_non_existent

Stack Traces | 0.029s run time

self = <unittest.case._Outcome object at 0x7fa113129b70>
test_case = <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_enroll_link_non_existent>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.13.5........./x64/lib/python3.13/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_enroll_link_non_existent>
result = <TestCaseFunction test_unauthenticated_enroll_link_non_existent>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
                    self._callSetUp()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.13.5........./x64/lib/python3.13/unittest/case.py:651: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_enroll_link_non_existent>
method = <bound method TestSourceFlowManager.test_unauthenticated_enroll_link_non_existent of <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_enroll_link_non_existent>>

    def _callTestMethod(self, method):
>       if method() is not None:

.../hostedtoolcache/Python/3.13.5........./x64/lib/python3.13/unittest/case.py:606: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_enroll_link_non_existent>

    def test_unauthenticated_enroll_link_non_existent(self):
        """Test un-authenticated user enrolling (link on username), username doesn't exist"""
        self.source.user_matching_mode = SourceUserMatchingModes.USERNAME_LINK
    
>       flow_manager = OAuthSourceFlowManager(
            self.source,
            get_request("/", user=AnonymousUser()),
            self.identifier,
            {
                "info": {"username": "foo"},
            },
            {},
        )
E       TypeError: SourceFlowManager.__init__() takes 5 positional arguments but 6 were given

.../core/tests/test_source_flow_manager.py:202: TypeError

authentik.core.tests.test_source_flow_manager.TestSourceFlowManager::test_unauthenticated_auth

Stack Traces | 0.037s run time

self = <unittest.case._Outcome object at 0x7fa11239c360>
test_case = <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_auth>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.13.5........./x64/lib/python3.13/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_auth>
result = <TestCaseFunction test_unauthenticated_auth>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
                    self._callSetUp()
                if outcome.success:
                    outcome.expecting_failure = expecting_failure
                    with outcome.testPartExecutor(self):
>                       self._callTestMethod(testMethod)

.../hostedtoolcache/Python/3.13.5........./x64/lib/python3.13/unittest/case.py:651: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_auth>
method = <bound method TestSourceFlowManager.test_unauthenticated_auth of <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_auth>>

    def _callTestMethod(self, method):
>       if method() is not None:

.../hostedtoolcache/Python/3.13.5........./x64/lib/python3.13/unittest/case.py:606: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <authentik.core.tests.test_source_flow_manager.TestSourceFlowManager testMethod=test_unauthenticated_auth>

    def test_unauthenticated_auth(self):
        """Test un-authenticated user authenticating"""
        UserOAuthSourceConnection.objects.create(
            user=get_anonymous_user(), source=self.source, identifier=self.identifier
        )
        request = get_request("/", user=AnonymousUser())
>       flow_manager = OAuthSourceFlowManager(
            self.source, request, self.identifier, {"info": {}}, {}
        )
E       TypeError: SourceFlowManager.__init__() takes 5 positional arguments but 6 were given

.../core/tests/test_source_flow_manager.py:57: TypeError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}