4102: Ensured dynamic role assignment is used in oauth

Author: cableman

backend/open_webui/models/roles.py

@@ -103,4 +103,16 @@ def delete_by_id(self, role_id: str) -> bool:
         except Exception:
             return False
 
+    def add_role_if_role_do_not_exists(self, role_name: str) -> Optional[RoleModel]:
+        # Check if role already exists
+        existing_role = self.get_role_by_name(role_name)
+        if existing_role:
+            return existing_role
+
+        # Role is allowed and doesn't exist, so create it
+        try:
+            return self.insert_new_role(name=role_name)
+        except Exception as e:
+            return None
+
 Roles = RolesTable()

backend/open_webui/utils/oauth.py

@@ -15,6 +15,7 @@
 
 from open_webui.models.auths import Auths
 from open_webui.models.users import Users
+from open_webui.models.roles import Roles
 from open_webui.models.groups import Groups, GroupModel, GroupUpdateForm
 from open_webui.config import (
     DEFAULT_USER_ROLE,
@@ -79,6 +80,17 @@ def __init__(self, app):
     def get_client(self, provider_name):
         return self.oauth.create_client(provider_name)
 
+    def find_first_role_match(self, oauth_roles, allowed_roles):
+        # Convert to sets for more efficient lookup if lists are large
+        oauth_roles_set = set(oauth_roles)
+        allowed_roles_set = set(allowed_roles)
+
+        # Find the intersection of the two sets
+        matching_roles = oauth_roles_set.intersection(allowed_roles_set)
+
+        # Return the first matching role if any matches found
+        return next(iter(matching_roles), None)
+
     def get_user_role(self, user, user_data):
         if user and Users.get_num_users() == 1:
             # If the user is the only user, assign the role "admin" - actually repairs role for single user on login
@@ -117,8 +129,15 @@ def get_user_role(self, user, user_data):
                 for allowed_role in oauth_allowed_roles:
                     # If the user has any of the allowed roles, assign the role "user"
                     if allowed_role in oauth_roles:
-                        log.debug("Assigned user the user role")
-                        role = "user"
+                        first_match = self.find_first_role_match(oauth_roles, oauth_allowed_roles)
+                        if first_match:
+                            Roles.add_role_if_role_do_not_exists(first_match)
+                            role = first_match
+                        else:
+                            # Fallback to role user.
+                            role = "user"
+
+                        log.debug(f"Assigned user the {role} role")
                         break
                 for admin_role in oauth_admin_roles:
                     # If the user has any of the admin roles, assign the role "admin"