fix: handle non-UUID sub claim in auth sync

Authentik JWT sub claim may not be a valid UUID.
Now derives a deterministic UUID from the sub string if parsing fails.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Jeroen Schweitzer
2026-01-04 17:45:05 +01:00
co-authored by Claude Opus 4.5
parent dd0997679f
commit dd5b794de4
2 changed files with 102 additions and 8 deletions
+75 -1
View File
@@ -102,7 +102,13 @@ class AuthService:
Returns:
Tuple of (User, is_new_user)
"""
authentik_id = uuid.UUID(token_info.sub)
# Parse authentik_id - may be UUID or other format
try:
authentik_id = uuid.UUID(token_info.sub)
except ValueError:
# If sub is not a valid UUID, derive one deterministically
logger.warning(f"sub claim is not a UUID: {token_info.sub}, deriving UUID")
authentik_id = uuid.uuid5(uuid.NAMESPACE_OID, token_info.sub)
# Try to find existing user
stmt = (
@@ -144,6 +150,74 @@ class AuthService:
await self.session.flush()
return user, is_new
async def sync_user_from_claims(
self,
authentik_id: uuid.UUID,
email: str,
name: str,
groups: list[str],
avatar_url: Optional[str] = None,
) -> tuple[User, bool]:
"""
Create or update user from forward auth claims (NPM X-authentik-* headers)
This is similar to sync_user() but works with raw claims instead of
TokenInfoSchema. Used for auto-syncing users on first web login via NPM.
Args:
authentik_id: The Authentik user UUID (from X-authentik-uid)
email: User email (from X-authentik-email)
name: User display name (from X-authentik-name)
groups: List of group names (from X-authentik-groups)
avatar_url: Optional avatar URL
Returns:
Tuple of (User, is_new_user)
"""
# Try to find existing user
stmt = (
select(User)
.options(selectinload(User.roles), selectinload(User.preferences))
.where(User.authentik_id == authentik_id)
)
result = await self.session.execute(stmt)
user = result.scalar_one_or_none()
is_new = user is None
if is_new:
# Create new user
user = User(
authentik_id=authentik_id,
email=email,
name=name or email,
avatar_url=avatar_url,
last_login=datetime.now(timezone.utc),
)
self.session.add(user)
await self.session.flush() # Get the user ID
# Create default preferences
preferences = UserPreferences(user_id=user.id)
self.session.add(preferences)
logger.info(f"Created new user from forward auth: {email}")
else:
# Update existing user
user.email = email
user.name = name or email
if avatar_url:
user.avatar_url = avatar_url
user.last_login = datetime.now(timezone.utc)
logger.info(f"Updated existing user from forward auth: {email}")
# Sync roles from groups
await self.sync_roles(user, groups)
await self.session.flush()
return user, is_new
async def sync_roles(self, user: User, group_names: list[str]) -> list[Role]:
"""
Synchronize user roles from Authentik groups via group_roles mapping