mirror of
https://github.com/open-webui/open-webui
synced 2024-11-06 16:59:42 +00:00
262 lines
11 KiB
Python
262 lines
11 KiB
Python
import base64
|
|
import logging
|
|
import mimetypes
|
|
import uuid
|
|
|
|
import aiohttp
|
|
from authlib.integrations.starlette_client import OAuth
|
|
from authlib.oidc.core import UserInfo
|
|
from fastapi import (
|
|
HTTPException,
|
|
status,
|
|
)
|
|
from starlette.responses import RedirectResponse
|
|
|
|
from open_webui.apps.webui.models.auths import Auths
|
|
from open_webui.apps.webui.models.users import Users
|
|
from open_webui.config import (
|
|
DEFAULT_USER_ROLE,
|
|
ENABLE_OAUTH_SIGNUP,
|
|
OAUTH_MERGE_ACCOUNTS_BY_EMAIL,
|
|
OAUTH_PROVIDERS,
|
|
ENABLE_OAUTH_ROLE_MANAGEMENT,
|
|
OAUTH_ROLES_CLAIM,
|
|
OAUTH_EMAIL_CLAIM,
|
|
OAUTH_PICTURE_CLAIM,
|
|
OAUTH_USERNAME_CLAIM,
|
|
OAUTH_ALLOWED_ROLES,
|
|
OAUTH_ADMIN_ROLES,
|
|
WEBHOOK_URL,
|
|
JWT_EXPIRES_IN,
|
|
AppConfig,
|
|
)
|
|
from open_webui.constants import ERROR_MESSAGES
|
|
from open_webui.env import WEBUI_SESSION_COOKIE_SAME_SITE, WEBUI_SESSION_COOKIE_SECURE
|
|
from open_webui.utils.misc import parse_duration
|
|
from open_webui.utils.utils import get_password_hash, create_token
|
|
from open_webui.utils.webhook import post_webhook
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
auth_manager_config = AppConfig()
|
|
auth_manager_config.DEFAULT_USER_ROLE = DEFAULT_USER_ROLE
|
|
auth_manager_config.ENABLE_OAUTH_SIGNUP = ENABLE_OAUTH_SIGNUP
|
|
auth_manager_config.OAUTH_MERGE_ACCOUNTS_BY_EMAIL = OAUTH_MERGE_ACCOUNTS_BY_EMAIL
|
|
auth_manager_config.ENABLE_OAUTH_ROLE_MANAGEMENT = ENABLE_OAUTH_ROLE_MANAGEMENT
|
|
auth_manager_config.OAUTH_ROLES_CLAIM = OAUTH_ROLES_CLAIM
|
|
auth_manager_config.OAUTH_EMAIL_CLAIM = OAUTH_EMAIL_CLAIM
|
|
auth_manager_config.OAUTH_PICTURE_CLAIM = OAUTH_PICTURE_CLAIM
|
|
auth_manager_config.OAUTH_USERNAME_CLAIM = OAUTH_USERNAME_CLAIM
|
|
auth_manager_config.OAUTH_ALLOWED_ROLES = OAUTH_ALLOWED_ROLES
|
|
auth_manager_config.OAUTH_ADMIN_ROLES = OAUTH_ADMIN_ROLES
|
|
auth_manager_config.WEBHOOK_URL = WEBHOOK_URL
|
|
auth_manager_config.JWT_EXPIRES_IN = JWT_EXPIRES_IN
|
|
|
|
|
|
class OAuthManager:
|
|
def __init__(self):
|
|
self.oauth = OAuth()
|
|
for provider_name, provider_config in OAUTH_PROVIDERS.items():
|
|
self.oauth.register(
|
|
name=provider_name,
|
|
client_id=provider_config["client_id"],
|
|
client_secret=provider_config["client_secret"],
|
|
server_metadata_url=provider_config["server_metadata_url"],
|
|
client_kwargs={
|
|
"scope": provider_config["scope"],
|
|
},
|
|
redirect_uri=provider_config["redirect_uri"],
|
|
)
|
|
|
|
def get_client(self, provider_name):
|
|
return self.oauth.create_client(provider_name)
|
|
|
|
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
|
|
return "admin"
|
|
if not user and Users.get_num_users() == 0:
|
|
# If there are no users, assign the role "admin", as the first user will be an admin
|
|
return "admin"
|
|
|
|
if auth_manager_config.ENABLE_OAUTH_ROLE_MANAGEMENT:
|
|
oauth_claim = auth_manager_config.OAUTH_ROLES_CLAIM
|
|
oauth_allowed_roles = auth_manager_config.OAUTH_ALLOWED_ROLES
|
|
oauth_admin_roles = auth_manager_config.OAUTH_ADMIN_ROLES
|
|
oauth_roles = None
|
|
role = "pending" # Default/fallback role if no matching roles are found
|
|
|
|
# Next block extracts the roles from the user data, accepting nested claims of any depth
|
|
if oauth_claim and oauth_allowed_roles and oauth_admin_roles:
|
|
claim_data = user_data
|
|
nested_claims = oauth_claim.split(".")
|
|
for nested_claim in nested_claims:
|
|
claim_data = claim_data.get(nested_claim, {})
|
|
oauth_roles = claim_data if isinstance(claim_data, list) else None
|
|
|
|
# If any roles are found, check if they match the allowed or admin roles
|
|
if oauth_roles:
|
|
# If role management is enabled, and matching roles are provided, use the roles
|
|
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:
|
|
role = "user"
|
|
break
|
|
for admin_role in oauth_admin_roles:
|
|
# If the user has any of the admin roles, assign the role "admin"
|
|
if admin_role in oauth_roles:
|
|
role = "admin"
|
|
break
|
|
else:
|
|
if not user:
|
|
# If role management is disabled, use the default role for new users
|
|
role = auth_manager_config.DEFAULT_USER_ROLE
|
|
else:
|
|
# If role management is disabled, use the existing role for existing users
|
|
role = user.role
|
|
|
|
return role
|
|
|
|
async def handle_login(self, provider, request):
|
|
if provider not in OAUTH_PROVIDERS:
|
|
raise HTTPException(404)
|
|
# If the provider has a custom redirect URL, use that, otherwise automatically generate one
|
|
redirect_uri = OAUTH_PROVIDERS[provider].get("redirect_uri") or request.url_for(
|
|
"oauth_callback", provider=provider
|
|
)
|
|
client = self.get_client(provider)
|
|
if client is None:
|
|
raise HTTPException(404)
|
|
return await client.authorize_redirect(request, redirect_uri)
|
|
|
|
async def handle_callback(self, provider, request, response):
|
|
if provider not in OAUTH_PROVIDERS:
|
|
raise HTTPException(404)
|
|
client = self.get_client(provider)
|
|
try:
|
|
token = await client.authorize_access_token(request)
|
|
except Exception as e:
|
|
log.warning(f"OAuth callback error: {e}")
|
|
raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_CRED)
|
|
user_data: UserInfo = token["userinfo"]
|
|
if not user_data:
|
|
user_data: UserInfo = await client.userinfo(token=token)
|
|
if not user_data:
|
|
log.warning(f"OAuth callback failed, user data is missing: {token}")
|
|
raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_CRED)
|
|
|
|
sub = user_data.get("sub")
|
|
if not sub:
|
|
log.warning(f"OAuth callback failed, sub is missing: {user_data}")
|
|
raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_CRED)
|
|
provider_sub = f"{provider}@{sub}"
|
|
email_claim = auth_manager_config.OAUTH_EMAIL_CLAIM
|
|
email = user_data.get(email_claim, "").lower()
|
|
# We currently mandate that email addresses are provided
|
|
if not email:
|
|
log.warning(f"OAuth callback failed, email is missing: {user_data}")
|
|
raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_CRED)
|
|
|
|
# Check if the user exists
|
|
user = Users.get_user_by_oauth_sub(provider_sub)
|
|
|
|
if not user:
|
|
# If the user does not exist, check if merging is enabled
|
|
if auth_manager_config.OAUTH_MERGE_ACCOUNTS_BY_EMAIL.value:
|
|
# Check if the user exists by email
|
|
user = Users.get_user_by_email(email)
|
|
if user:
|
|
# Update the user with the new oauth sub
|
|
Users.update_user_oauth_sub_by_id(user.id, provider_sub)
|
|
|
|
if user:
|
|
determined_role = self.get_user_role(user, user_data)
|
|
if user.role != determined_role:
|
|
Users.update_user_role_by_id(user.id, determined_role)
|
|
|
|
if not user:
|
|
# If the user does not exist, check if signups are enabled
|
|
if auth_manager_config.ENABLE_OAUTH_SIGNUP.value:
|
|
# Check if an existing user with the same email already exists
|
|
existing_user = Users.get_user_by_email(
|
|
user_data.get("email", "").lower()
|
|
)
|
|
if existing_user:
|
|
raise HTTPException(400, detail=ERROR_MESSAGES.EMAIL_TAKEN)
|
|
|
|
picture_claim = auth_manager_config.OAUTH_PICTURE_CLAIM
|
|
picture_url = user_data.get(picture_claim, "")
|
|
if picture_url:
|
|
# Download the profile image into a base64 string
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(picture_url) as resp:
|
|
picture = await resp.read()
|
|
base64_encoded_picture = base64.b64encode(
|
|
picture
|
|
).decode("utf-8")
|
|
guessed_mime_type = mimetypes.guess_type(picture_url)[0]
|
|
if guessed_mime_type is None:
|
|
# assume JPG, browsers are tolerant enough of image formats
|
|
guessed_mime_type = "image/jpeg"
|
|
picture_url = f"data:{guessed_mime_type};base64,{base64_encoded_picture}"
|
|
except Exception as e:
|
|
log.error(
|
|
f"Error downloading profile image '{picture_url}': {e}"
|
|
)
|
|
picture_url = ""
|
|
if not picture_url:
|
|
picture_url = "/user.png"
|
|
username_claim = auth_manager_config.OAUTH_USERNAME_CLAIM
|
|
|
|
role = self.get_user_role(None, user_data)
|
|
|
|
user = Auths.insert_new_auth(
|
|
email=email,
|
|
password=get_password_hash(
|
|
str(uuid.uuid4())
|
|
), # Random password, not used
|
|
name=user_data.get(username_claim, "User"),
|
|
profile_image_url=picture_url,
|
|
role=role,
|
|
oauth_sub=provider_sub,
|
|
)
|
|
|
|
if auth_manager_config.WEBHOOK_URL:
|
|
post_webhook(
|
|
auth_manager_config.WEBHOOK_URL,
|
|
auth_manager_config.WEBHOOK_MESSAGES.USER_SIGNUP(user.name),
|
|
{
|
|
"action": "signup",
|
|
"message": auth_manager_config.WEBHOOK_MESSAGES.USER_SIGNUP(
|
|
user.name
|
|
),
|
|
"user": user.model_dump_json(exclude_none=True),
|
|
},
|
|
)
|
|
else:
|
|
raise HTTPException(
|
|
status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.ACCESS_PROHIBITED
|
|
)
|
|
|
|
jwt_token = create_token(
|
|
data={"id": user.id},
|
|
expires_delta=parse_duration(auth_manager_config.JWT_EXPIRES_IN),
|
|
)
|
|
|
|
# Set the cookie token
|
|
response.set_cookie(
|
|
key="token",
|
|
value=jwt_token,
|
|
httponly=True, # Ensures the cookie is not accessible via JavaScript
|
|
samesite=WEBUI_SESSION_COOKIE_SAME_SITE,
|
|
secure=WEBUI_SESSION_COOKIE_SECURE,
|
|
)
|
|
|
|
# Redirect back to the frontend with the JWT token
|
|
redirect_url = f"{request.base_url}auth#token={jwt_token}"
|
|
return RedirectResponse(url=redirect_url)
|
|
|
|
|
|
oauth_manager = OAuthManager()
|