@arizeai/phoenix-mcp

"""Integration tests for LDAP authentication. These tests use a real in-process mock LDAP server that implements the LDAP protocol, similar to how OIDC tests use _OIDCServer. The mock server runs in a separate thread and listens on a dynamically allocated port. """ from __future__ import annotations from dataclasses import dataclass from secrets import token_hex from typing import Optional from phoenix.server.api.input_types.UserRoleInput import UserRoleInput from tests.integration._mock_ldap_server import _LDAPServer from .._helpers import ( _AppInfo, _delete_users, _httpx_client, _list_users, _User, ) # Test constants _ADMIN_GROUP = "cn=admins,ou=groups,dc=example,dc=com" _MEMBER_GROUP = "cn=members,ou=groups,dc=example,dc=com" _DEFAULT_PASSWORD = "password123" @dataclass class LDAPTestUser: """Test user specification for LDAP tests.""" username: str password: str email: str display_name: str groups: list[str] expected_role: UserRoleInput def _create_test_user( ldap_server: _LDAPServer, suffix: str, base_username: str, role: UserRoleInput, groups: Optional[list[str]] = None, ) -> LDAPTestUser: """Create a test user with unique identifiers for isolation. Args: ldap_server: Mock LDAP server to add user to suffix: Unique suffix (e.g., token_hex(4)) base_username: Base username before suffix role: Expected Phoenix role groups: LDAP groups (defaults to role-appropriate groups) Returns: LDAPTestUser specification """ if groups is None: role_to_groups = { UserRoleInput.ADMIN: [_ADMIN_GROUP], UserRoleInput.MEMBER: [_MEMBER_GROUP], UserRoleInput.VIEWER: [], # Wildcard } groups = role_to_groups.get(role, []) username = f"{base_username}_{suffix}" email = f"{base_username}_{suffix}@example.com" display_name = base_username.replace("_", " ").title() ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email=email, display_name=display_name, groups=groups, ) return LDAPTestUser( username=username, password=_DEFAULT_PASSWORD, email=email, display_name=display_name, groups=groups, expected_role=role, ) def _get_user_by_email(app: _AppInfo, email: str) -> Optional[_User]: """Get user by email from the user list.""" users = {u.profile.email: u for u in _list_users(app, app.admin_secret)} return users.get(email) def _verify_ldap_login_success( status_code: int, access_token: Optional[str], refresh_token: Optional[str], ) -> None: """Verify LDAP login was successful and tokens were issued. Checks: - HTTP 204 status - Access token present and non-empty - Refresh token present and non-empty """ assert status_code == 204, f"Expected 204, got {status_code}" assert access_token is not None, "Access token must be present" assert refresh_token is not None, "Refresh token must be present" assert len(access_token) > 10, f"Access token too short: {len(access_token)} chars" assert len(refresh_token) > 10, f"Refresh token too short: {len(refresh_token)} chars" def _verify_user_created( app: _AppInfo, test_user: LDAPTestUser, expected_username: Optional[str] = None, ) -> _User: """Verify user was created in Phoenix DB with correct attributes. Args: app: Application info test_user: Test user specification expected_username: Expected username (defaults to test_user.display_name) Returns: The created user object """ user = _get_user_by_email(app, test_user.email) assert user is not None, f"User not found: {test_user.email}" assert user.role == test_user.expected_role, ( f"Expected role {test_user.expected_role}, got {user.role}" ) expected_username = expected_username or test_user.display_name assert user.profile.username == expected_username, ( f"Expected username '{expected_username}', got '{user.profile.username}'" ) return user def _ldap_login( app: _AppInfo, username: str, password: str ) -> tuple[int, Optional[str], Optional[str]]: """Perform LDAP login and return response details. Returns: Tuple of (status_code, access_token, refresh_token) """ client = _httpx_client(app) response = client.post( "/auth/ldap/login", json={"username": username, "password": password}, ) return ( response.status_code, response.cookies.get("phoenix-access-token"), response.cookies.get("phoenix-refresh-token"), ) class TestLDAPAuthentication: """Test LDAP authentication - core flows and security.""" async def test_authentication_and_role_mapping( self, _app: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test successful login assigns role from group membership.""" suffix = token_hex(4) # Admin user (in admins group) admin = _create_test_user(_ldap_server, suffix, "admin", UserRoleInput.ADMIN) status, access_token, refresh_token = _ldap_login(_app, admin.username, admin.password) _verify_ldap_login_success(status, access_token, refresh_token) admin_user = _verify_user_created(_app, admin) # Member user (in members group) member = _create_test_user(_ldap_server, suffix, "member", UserRoleInput.MEMBER) status, access_token, refresh_token = _ldap_login(_app, member.username, member.password) _verify_ldap_login_success(status, access_token, refresh_token) member_user = _verify_user_created(_app, member) # Viewer user (no groups → wildcard) viewer = _create_test_user(_ldap_server, suffix, "viewer", UserRoleInput.VIEWER, groups=[]) status, access_token, refresh_token = _ldap_login(_app, viewer.username, viewer.password) _verify_ldap_login_success(status, access_token, refresh_token) viewer_user = _verify_user_created(_app, viewer) _delete_users( _app, _app.admin_secret, users=[admin_user.gid, member_user.gid, viewer_user.gid] ) async def test_invalid_credentials_rejected( self, _app: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test login fails for wrong password, nonexistent user, and empty credentials.""" user = _create_test_user(_ldap_server, token_hex(4), "user", UserRoleInput.ADMIN) # Wrong password assert _ldap_login(_app, user.username, "wrong")[0] == 401 # Nonexistent user assert _ldap_login(_app, "nonexistent", "pass")[0] == 401 # Empty credentials assert _ldap_login(_app, "", "")[0] == 401 async def test_role_syncs_on_subsequent_login( self, _app: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test role updates from LDAP groups but username remains stable.""" suffix = token_hex(4) username, email = f"sync_{suffix}", f"sync_{suffix}@example.com" # First login with members group (mapped to MEMBER) _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email=email, display_name="Original Name", groups=[_MEMBER_GROUP], ) status, _, _ = _ldap_login(_app, username, _DEFAULT_PASSWORD) assert status == 204 user = _get_user_by_email(_app, email) assert user is not None and user.role == UserRoleInput.MEMBER # LDAP changes: promoted to admin, display name changed _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email=email, display_name="New Name", groups=[_ADMIN_GROUP], ) status, _, _ = _ldap_login(_app, username, _DEFAULT_PASSWORD) assert status == 204 # Verify: role updated, username stable updated = _get_user_by_email(_app, email) assert updated is not None assert updated.gid == user.gid # Same user assert updated.role == UserRoleInput.ADMIN # Role synced assert updated.profile.username == "Original Name" # Username stable _delete_users(_app, _app.admin_secret, users=[user.gid]) async def test_injection_prevention(self, _app: _AppInfo, _ldap_server: _LDAPServer) -> None: """Test LDAP injection attempts are rejected.""" payloads = [ "*", # Wildcard "admin*", # Wildcard suffix "*(objectClass=*)", # Filter injection "admin)(|(objectClass=*", # Filter escape "admin\x00injected", # Null byte injection "admin\ninjected", # Newline injection "admin\r\ninjected", # CRLF injection ")(cn=*", # DN injection ] for payload in payloads: assert _ldap_login(_app, payload, _DEFAULT_PASSWORD)[0] == 401 async def test_unicode_credentials(self, _app: _AppInfo, _ldap_server: _LDAPServer) -> None: """Test login with Unicode username/password.""" suffix = token_hex(4) email = f"unicode_{suffix}@example.com" _ldap_server.add_user( username="用户名", password="密码123", email=email, display_name="Unicode User", groups=[_MEMBER_GROUP], ) status, access_token, refresh_token = _ldap_login(_app, "用户名", "密码123") _verify_ldap_login_success(status, access_token, refresh_token) user = _get_user_by_email(_app, email) assert user is not None _delete_users(_app, _app.admin_secret, users=[user.gid]) async def test_special_characters_in_password( self, _app: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test login with special characters in password (quotes, backslashes, etc.).""" suffix = token_hex(4) email = f"special_{suffix}@example.com" special_password = r'p@ss"word\with\'special<chars>&more!' _ldap_server.add_user( username=f"special_{suffix}", password=special_password, email=email, display_name="Special User", groups=[_MEMBER_GROUP], ) status, access_token, refresh_token = _ldap_login( _app, f"special_{suffix}", special_password ) _verify_ldap_login_success(status, access_token, refresh_token) user = _get_user_by_email(_app, email) assert user is not None _delete_users(_app, _app.admin_secret, users=[user.gid]) async def test_missing_email_rejected(self, _app: _AppInfo, _ldap_server: _LDAPServer) -> None: """Test login fails when LDAP user has no email.""" suffix = token_hex(4) _ldap_server.add_user( username=f"noemail_{suffix}", password=_DEFAULT_PASSWORD, email="", display_name="No Email", groups=[_ADMIN_GROUP], ) assert _ldap_login(_app, f"noemail_{suffix}", _DEFAULT_PASSWORD)[0] == 401 async def test_missing_display_name_uses_fallback( self, _app: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test missing displayName falls back to email prefix.""" suffix = token_hex(4) email = f"noname_{suffix}@example.com" _ldap_server.add_user( username=f"noname_{suffix}", password=_DEFAULT_PASSWORD, email=email, display_name="", groups=[_MEMBER_GROUP], ) status, _, _ = _ldap_login(_app, f"noname_{suffix}", _DEFAULT_PASSWORD) assert status == 204 user = _get_user_by_email(_app, email) assert user is not None assert user.profile.username == f"noname_{suffix}" # Fallback to email prefix _delete_users(_app, _app.admin_secret, users=[user.gid]) async def test_multiple_groups_uses_first_match( self, _app: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test user in multiple groups gets role from first matching mapping. group_role_mappings is evaluated in order: ADMIN, MEMBER, VIEWER (wildcard). User is in both MEMBER and ADMIN groups, but ADMIN mapping is checked first. """ suffix = token_hex(4) email = f"multi_{suffix}@example.com" _ldap_server.add_user( username=f"multi_{suffix}", password=_DEFAULT_PASSWORD, email=email, display_name="Multi", groups=[_MEMBER_GROUP, _ADMIN_GROUP], # Both groups ) status, _, _ = _ldap_login(_app, f"multi_{suffix}", _DEFAULT_PASSWORD) assert status == 204 user = _get_user_by_email(_app, email) assert user is not None assert user.role == UserRoleInput.ADMIN # ADMIN mapping evaluated before MEMBER _delete_users(_app, _app.admin_secret, users=[user.gid]) class TestLDAPUserIdentificationStrategies: """Test LDAP user identification strategies. Phoenix supports two modes: 1. Simple Mode (default): Email is the identifier. Email changes create new users. 2. Enterprise Mode (PHOENIX_LDAP_ATTR_UNIQUE_ID): Stable ID is the identifier. Email changes preserve identity. """ def test_email_change_creates_new_user_without_unique_id( self, _app: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Simple mode: email change creates a new user (email is the identifier).""" suffix = token_hex(4) username = f"simple_{suffix}" email_v1 = f"simple_v1_{suffix}@example.com" email_v2 = f"simple_v2_{suffix}@example.com" # First login with email_v1 _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email=email_v1, display_name="User", groups=[_ADMIN_GROUP], ) status, access_token, refresh_token = _ldap_login(_app, username, _DEFAULT_PASSWORD) _verify_ldap_login_success(status, access_token, refresh_token) user_v1 = _get_user_by_email(_app, email_v1) assert user_v1 is not None # Email changes in LDAP, login again _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email=email_v2, display_name="User", groups=[_ADMIN_GROUP], ) status, access_token, refresh_token = _ldap_login(_app, username, _DEFAULT_PASSWORD) _verify_ldap_login_success(status, access_token, refresh_token) # Result: NEW user created (email is the identifier) user_v2 = _get_user_by_email(_app, email_v2) assert user_v2 is not None assert user_v2.gid != user_v1.gid, "Different user when email changes" # Cleanup both users _delete_users(_app, _app.admin_secret, users=[user_v1.gid, user_v2.gid]) def test_email_change_preserves_identity_with_unique_id( self, _app_ldap_unique_id: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Enterprise mode: email change preserves identity (unique_id is the identifier). Also tests migration: admin pre-provisions user, first login migrates to unique_id, subsequent email change still preserves identity. """ app = _app_ldap_unique_id suffix = token_hex(4) username = f"enterprise_{suffix}" email_v1 = f"enterprise_v1_{suffix}@example.com" email_v2 = f"enterprise_v2_{suffix}@example.com" # Admin pre-provisions user (no unique_id in DB yet) graphql_client = _httpx_client(app, app.admin_secret) username = f"Pre-Provisioned {suffix}" response = graphql_client.post( "/graphql", json={ "query": """ mutation($email: String!, $username: String!, $role: UserRoleInput!) { createUser(input: { email: $email, username: $username, role: $role, authMethod: LDAP }) { user { id } } } """, "variables": {"email": email_v1, "username": username, "role": "MEMBER"}, }, ) assert response.status_code == 200 response_json = response.json() assert not response_json.get("errors") # First login (migrates pre-provisioned user to unique_id) _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email=email_v1, display_name=f"User {token_hex(4)}", groups=[_ADMIN_GROUP], ) status, access_token, refresh_token = _ldap_login(app, username, _DEFAULT_PASSWORD) _verify_ldap_login_success(status, access_token, refresh_token) user_v1 = _get_user_by_email(app, email_v1) assert user_v1 is not None assert user_v1.role == UserRoleInput.ADMIN, "Role updated from LDAP groups" assert user_v1.username == username, "Username stable from pre-provisioning" # Email changes in LDAP, login again _ldap_server.add_user( username=username, # Same username = same entryUUID password=_DEFAULT_PASSWORD, email=email_v2, display_name=f"User {token_hex(4)}", groups=[_MEMBER_GROUP], ) status, access_token, refresh_token = _ldap_login(app, username, _DEFAULT_PASSWORD) _verify_ldap_login_success(status, access_token, refresh_token) # Result: SAME user (unique_id is the identifier) user_v2 = _get_user_by_email(app, email_v2) assert user_v2 is not None assert user_v2.gid == user_v1.gid, "Same user when unique_id is configured" assert user_v2.role == UserRoleInput.MEMBER assert user_v2.username == username # Old email no longer exists assert _get_user_by_email(app, email_v1) is None # Cleanup _delete_users(app, app.admin_secret, users=[user_v2.gid]) class TestLDAPGraphQLIntegration: """Test GraphQL integration for LDAP users.""" async def test_ldap_user_graphql_auth_method( self, _app: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test that GraphQL correctly exposes LDAP users with authMethod='LDAP'. Verifies the GraphQL resolver translates the database storage (OAuth2 with marker) to the semantic AuthMethod.LDAP for the frontend. """ test_user = _create_test_user( _ldap_server, token_hex(4), "graphql_user", UserRoleInput.ADMIN ) # Step 1: Create LDAP user via login status, access_token, refresh_token = _ldap_login( _app, test_user.username, test_user.password ) _verify_ldap_login_success(status, access_token, refresh_token) # Step 2: Query GraphQL to verify authMethod is 'LDAP' graphql_query = """ query { users { edges { user: node { email authMethod } } } } """ graphql_client = _httpx_client(_app, _app.admin_secret) graphql_response = graphql_client.post( "/graphql", json={"query": graphql_query}, ) assert graphql_response.status_code == 200 graphql_data = graphql_response.json() assert not graphql_data.get("errors"), graphql_data.get("errors") data = graphql_data # Find our LDAP user in the response users = data["data"]["users"]["edges"] ldap_user_data = next((u for u in users if u["user"]["email"] == test_user.email), None) assert ldap_user_data is not None, f"User {test_user.email} not found in GraphQL response" # The key assertion: GraphQL should expose authMethod as 'LDAP' (not 'OAUTH2') assert ldap_user_data["user"]["authMethod"] == "LDAP", ( f"Expected authMethod='LDAP', got '{ldap_user_data['user']['authMethod']}'" ) # Cleanup user = _get_user_by_email(_app, test_user.email) if user: _delete_users(_app, _app.admin_secret, users=[user.gid]) class TestLDAPSecurityIsolation: """Test that LDAP, OAuth2, and LOCAL auth methods are isolated from each other.""" async def test_local_user_protected_from_ldap_login( self, _app: _AppInfo, _ldap_server: _LDAPServer ) -> None: """LOCAL user cannot be hijacked via LDAP login with same email.""" suffix = token_hex(4) email = f"local_{suffix}@example.com" # Create LOCAL user graphql_client = _httpx_client(_app, _app.admin_secret) resp = graphql_client.post( "/graphql", json={ "query": """mutation($i: CreateUserInput!) { createUser(input: $i) { user { id } } }""", "variables": { "i": { "email": email, "username": "Local", "role": "MEMBER", "authMethod": "LOCAL", "password": "pass123", } }, }, ) local_user_id = resp.json()["data"]["createUser"]["user"]["id"] # Add same email to LDAP _ldap_server.add_user( username=f"local_{suffix}", password="ldappass", email=email, display_name="LDAP", groups=[_ADMIN_GROUP], ) # LDAP login should fail (LOCAL user protected) assert _ldap_login(_app, f"local_{suffix}", "ldappass")[0] == 401 _delete_users(_app, _app.admin_secret, users=[local_user_id]) async def test_ldap_user_protected_from_password_login( self, _app: _AppInfo, _ldap_server: _LDAPServer ) -> None: """LDAP user cannot login via password endpoint (no password_hash).""" user = _create_test_user(_ldap_server, token_hex(4), "ldap", UserRoleInput.ADMIN) status, _, _ = _ldap_login(_app, user.username, user.password) assert status == 204 # Password login should fail client = _httpx_client(_app) resp = client.post("/auth/login", json={"email": user.email, "password": user.password}) assert resp.status_code == 401 db_user = _get_user_by_email(_app, user.email) assert db_user is not None _delete_users(_app, _app.admin_secret, users=[db_user.gid]) async def test_oauth2_user_protected_from_ldap_login( self, _app: _AppInfo, _ldap_server: _LDAPServer ) -> None: """OAuth2 user cannot be hijacked via LDAP login with same email.""" suffix = token_hex(4) email = f"oauth_{suffix}@example.com" # Create OAuth2 user via REST API client = _httpx_client(_app, _app.admin_secret) resp = client.post( "/v1/users", json={ "user": { "email": email, "username": f"oauth_{suffix}", "role": "VIEWER", "auth_method": "OAUTH2", "oauth2_client_id": "google", "oauth2_user_id": f"google-{suffix}", }, "send_welcome_email": False, }, ) assert resp.status_code == 201 oauth_user_id = resp.json()["data"]["id"] # Add same email to LDAP _ldap_server.add_user( username=f"oauth_{suffix}", password=_DEFAULT_PASSWORD, email=email, display_name="LDAP", groups=[_ADMIN_GROUP], ) # LDAP login should fail (OAuth2 user protected) assert _ldap_login(_app, f"oauth_{suffix}", _DEFAULT_PASSWORD)[0] == 401 client.delete(f"/v1/users/{oauth_user_id}") class TestLDAPConfiguration: """Test LDAP configuration-specific behaviors.""" def test_ldap_allow_sign_up_false_with_email_lookup( self, _app_ldap_no_sign_up: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test email-based user lookup when allow_sign_up=false. Critical behavior for allow_sign_up=false: 1. Admin creates LDAP user with email "john@example.com" 2. User logs in via LDAP with any username 3. Phoenix gets email from LDAP authentication 4. Phoenix looks up user by email → finds user 5. Login succeeds and attributes synced from LDAP """ # Step 1: Admin pre-creates LDAP user (only needs email) suffix = token_hex(4) ldap_username = f"jdoe_{suffix}" # What user will login with email = f"john_{suffix}@example.com" # Admin creates user via GraphQL with wrong username graphql_client = _httpx_client(_app_ldap_no_sign_up, _app_ldap_no_sign_up.admin_secret) create_response = graphql_client.post( "/graphql", json={ "query": """ mutation CreateUser($input: CreateUserInput!) { createUser(input: $input) { user { id email username authMethod } } } """, "variables": { "input": { "email": email, "username": "John Doe", # Display name "role": "MEMBER", "authMethod": "LDAP", }, }, }, ) assert create_response.status_code == 200 create_json = create_response.json() assert not create_json.get("errors"), create_json.get("errors") user_data = create_json["data"]["createUser"]["user"] assert user_data["authMethod"] == "LDAP" created_user_gid = user_data["id"] # Step 2: Add user to LDAP with matching email _ldap_server.add_user( username=ldap_username, password=_DEFAULT_PASSWORD, email=email, display_name="John Smith, Ph.D.", groups=[_ADMIN_GROUP], ) # Step 3: User logs in via LDAP (email lookup finds pre-created user) status, access_token, refresh_token = _ldap_login( _app_ldap_no_sign_up, ldap_username, _DEFAULT_PASSWORD ) _verify_ldap_login_success(status, access_token, refresh_token) # Step 4: Verify username remains stable (not synced from LDAP) users = _list_users(_app_ldap_no_sign_up, _app_ldap_no_sign_up.admin_secret) updated_user = next((u for u in users if u.profile.email == email), None) assert updated_user is not None # Username stays stable from admin creation (prevents collisions on displayName changes) assert updated_user.profile.username == "John Doe" assert updated_user.role == UserRoleInput.ADMIN # Step 5: Verify subsequent logins work status, access_token, refresh_token = _ldap_login( _app_ldap_no_sign_up, ldap_username, _DEFAULT_PASSWORD ) _verify_ldap_login_success(status, access_token, refresh_token) _delete_users( _app_ldap_no_sign_up, _app_ldap_no_sign_up.admin_secret, users=[created_user_gid] ) class TestLDAPPosixGroupSearch: """Test LDAP group search (no memberOf attribute). Tests group lookup via search filter instead of reading memberOf from user entry. Uses (member=%s) where %s is replaced with the login username directly (no GROUP_SEARCH_FILTER_USER_ATTR configured). Note: For true POSIX RFC 2307 memberUid testing with GROUP_SEARCH_FILTER_USER_ATTR, see TestLDAPPosixMemberUidGroupSearch. """ def test_posix_role_from_group_search( self, _app_ldap_posix: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test role assignment via POSIX group search (no memberOf attribute).""" suffix = token_hex(4) username = f"posix_{suffix}" email = f"posix_{suffix}@example.com" # User without memberOf _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email=email, display_name="POSIX", groups=[], ) # Add to admins group with username as member _ldap_server.add_group(cn="admins", members=[username]) status, _, _ = _ldap_login(_app_ldap_posix, username, _DEFAULT_PASSWORD) assert status == 204 user = _get_user_by_email(_app_ldap_posix, email) assert user is not None assert user.role == UserRoleInput.ADMIN _delete_users(_app_ldap_posix, _app_ldap_posix.admin_secret, users=[user.gid]) def test_posix_wildcard_when_no_groups( self, _app_ldap_posix: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test wildcard role when user is in no groups.""" suffix = token_hex(4) username = f"posix_none_{suffix}" email = f"posix_none_{suffix}@example.com" _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email=email, display_name="No Groups", groups=[], ) # Don't add to any group status, _, _ = _ldap_login(_app_ldap_posix, username, _DEFAULT_PASSWORD) assert status == 204 user = _get_user_by_email(_app_ldap_posix, email) assert user is not None assert user.role == UserRoleInput.VIEWER # Wildcard _delete_users(_app_ldap_posix, _app_ldap_posix.admin_secret, users=[user.gid]) def test_posix_username_case_insensitive( self, _app_ldap_posix: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test username matching in group search is case-insensitive.""" suffix = token_hex(4) username = f"posix_case_{suffix}" email = f"posix_case_{suffix}@example.com" _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email=email, display_name="Case", groups=[], ) # Group has UPPERCASE username, user logs in with lowercase _ldap_server.add_group(cn="admins", members=[username.upper()]) status, _, _ = _ldap_login(_app_ldap_posix, username, _DEFAULT_PASSWORD) assert status == 204 user = _get_user_by_email(_app_ldap_posix, email) assert user is not None assert user.role == UserRoleInput.ADMIN # Should match despite case _delete_users(_app_ldap_posix, _app_ldap_posix.admin_secret, users=[user.gid]) class TestLDAPPosixMemberUidGroupSearch: """Test LDAP POSIX memberUid group search with GROUP_SEARCH_FILTER_USER_ATTR. This tests the code path where Phoenix must read a specific attribute (e.g., uid) from the user entry to substitute into the group search filter. Configuration: - GROUP_SEARCH_FILTER: (memberUid=%s) - GROUP_SEARCH_FILTER_USER_ATTR: uid The mock LDAP server respects the requested attributes list, so these tests properly catch bugs where required attributes (like uid) aren't being requested. This caught a real bug where group_search_filter_user_attr wasn't included in the LDAP search attribute list. """ def test_memberuid_role_from_group_search( self, _app_ldap_posix_memberuid: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test role assignment when GROUP_SEARCH_FILTER_USER_ATTR is configured.""" suffix = token_hex(4) username = f"memberuid_{suffix}" email = f"memberuid_{suffix}@example.com" _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email=email, display_name="MemberUid User", groups=[], # No memberOf - relies on group search ) # Add to admins group with username (not DN) _ldap_server.add_group(cn="admins", members=[username]) status, access_token, refresh_token = _ldap_login( _app_ldap_posix_memberuid, username, _DEFAULT_PASSWORD ) _verify_ldap_login_success(status, access_token, refresh_token) user = _get_user_by_email(_app_ldap_posix_memberuid, email) assert user is not None assert user.role == UserRoleInput.ADMIN, ( f"Expected ADMIN role from memberUid group search, got {user.role}. " "This indicates GROUP_SEARCH_FILTER_USER_ATTR is not working correctly." ) _delete_users( _app_ldap_posix_memberuid, _app_ldap_posix_memberuid.admin_secret, users=[user.gid] ) def test_memberuid_member_role( self, _app_ldap_posix_memberuid: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test MEMBER role assignment via memberUid group search.""" suffix = token_hex(4) username = f"memberuid_member_{suffix}" email = f"memberuid_member_{suffix}@example.com" _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email=email, display_name="Member User", groups=[], ) _ldap_server.add_group(cn="members", members=[username]) status, _, _ = _ldap_login(_app_ldap_posix_memberuid, username, _DEFAULT_PASSWORD) assert status == 204 user = _get_user_by_email(_app_ldap_posix_memberuid, email) assert user is not None assert user.role == UserRoleInput.MEMBER _delete_users( _app_ldap_posix_memberuid, _app_ldap_posix_memberuid.admin_secret, users=[user.gid] ) def test_memberuid_wildcard_when_no_groups( self, _app_ldap_posix_memberuid: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test wildcard role when user is in no groups (memberUid mode).""" suffix = token_hex(4) username = f"memberuid_none_{suffix}" email = f"memberuid_none_{suffix}@example.com" _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email=email, display_name="No Groups", groups=[], ) # Don't add to any group status, _, _ = _ldap_login(_app_ldap_posix_memberuid, username, _DEFAULT_PASSWORD) assert status == 204 user = _get_user_by_email(_app_ldap_posix_memberuid, email) assert user is not None assert user.role == UserRoleInput.VIEWER # Wildcard _delete_users( _app_ldap_posix_memberuid, _app_ldap_posix_memberuid.admin_secret, users=[user.gid] ) class TestLDAPNoEmailMode: """Test LDAP no-email mode (null email markers). When PHOENIX_LDAP_ATTR_EMAIL="" is configured, Phoenix generates null email markers for users instead of reading email from LDAP. Users are identified by their unique_id (entryUUID). This mode is useful for LDAP directories that don't have email attributes. """ def _get_user_with_username(self, app: _AppInfo, username: str) -> Optional[_User]: """Find user by checking for null email marker containing username hash.""" users = _list_users(app, app.admin_secret) for user in users: if user.profile.username == username: return user return None def test_login_creates_user_with_null_email_marker( self, _app_ldap_no_email: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test login in no-email mode creates user with null email marker.""" suffix = token_hex(4) username = f"noemail_{suffix}" display_name = f"No Email User {suffix}" _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email="ignored@example.com", # This email is ignored in no-email mode display_name=display_name, groups=[_ADMIN_GROUP], ) status, access_token, refresh_token = _ldap_login( _app_ldap_no_email, username, _DEFAULT_PASSWORD ) _verify_ldap_login_success(status, access_token, refresh_token) # Find the user - should have null email marker user = self._get_user_with_username(_app_ldap_no_email, display_name) assert user is not None, "User with null email marker should be created" assert user.profile.email == "", "User should have null email on first login" assert user.role == UserRoleInput.ADMIN _delete_users(_app_ldap_no_email, _app_ldap_no_email.admin_secret, users=[user.gid]) def test_subsequent_login_finds_same_user( self, _app_ldap_no_email: _AppInfo, _ldap_server: _LDAPServer ) -> None: """Test subsequent login in no-email mode finds the same user by unique_id.""" suffix = token_hex(4) username = f"noemail_same_{suffix}" display_name = f"Same User {suffix}" _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email="ignored@example.com", display_name=display_name, groups=[_MEMBER_GROUP], ) # First login status, access_token, refresh_token = _ldap_login( _app_ldap_no_email, username, _DEFAULT_PASSWORD ) _verify_ldap_login_success(status, access_token, refresh_token) user1 = self._get_user_with_username(_app_ldap_no_email, display_name) assert user1 is not None, "User with same username should be found on first login" assert user1.profile.email == "", "User should have null email on first login" assert user1.role is UserRoleInput.MEMBER, "User should have MEMBER role on first login" # Role changes in LDAP, login again _ldap_server.add_user( username=username, password=_DEFAULT_PASSWORD, email="ignored@example.com", display_name=display_name, groups=[_ADMIN_GROUP], ) # Second login - should find same user status, access_token, refresh_token = _ldap_login( _app_ldap_no_email, username, _DEFAULT_PASSWORD ) _verify_ldap_login_success(status, access_token, refresh_token) user2 = self._get_user_with_username(_app_ldap_no_email, display_name) assert user2 is not None, "User with same username should be found on subsequent login" assert user2.profile.email == "", "User should have null email on subsequent login" assert user2.role is UserRoleInput.ADMIN, "User should have ADMIN role on subsequent login" assert user2.gid == user1.gid, "Same user should be found on subsequent login" _delete_users(_app_ldap_no_email, _app_ldap_no_email.admin_secret, users=[user1.gid])