Merge pull request #1347 from cheahjs/feat/trusted-email-header

feat: allow authenticating with a trusted email header

backend/apps/web/main.py

@@ -20,6 +20,7 @@ from config import (
     ENABLE_SIGNUP,
     USER_PERMISSIONS,
     WEBHOOK_URL,
+    WEBUI_AUTH_TRUSTED_EMAIL_HEADER,
 )
 
 app = FastAPI()
@@ -34,7 +35,7 @@ app.state.DEFAULT_PROMPT_SUGGESTIONS = DEFAULT_PROMPT_SUGGESTIONS
 app.state.DEFAULT_USER_ROLE = DEFAULT_USER_ROLE
 app.state.USER_PERMISSIONS = USER_PERMISSIONS
 app.state.WEBHOOK_URL = WEBHOOK_URL
-
+app.state.AUTH_TRUSTED_EMAIL_HEADER = WEBUI_AUTH_TRUSTED_EMAIL_HEADER
 
 app.add_middleware(
     CORSMiddleware,

backend/apps/web/models/auths.py

@@ -123,6 +123,16 @@ class AuthsTable:
         except:
             return None
 
+    def authenticate_user_by_trusted_header(self, email: str) -> Optional[UserModel]:
+        log.info(f"authenticate_user_by_trusted_header: {email}")
+        try:
+            auth = Auth.get(Auth.email == email, Auth.active == True)
+            if auth:
+                user = Users.get_user_by_id(auth.id)
+                return user
+        except:
+            return None
+
     def update_user_password_by_id(self, id: str, new_password: str) -> bool:
         try:
             query = Auth.update(password=new_password).where(Auth.id == id)

backend/apps/web/routers/auths.py

@@ -29,6 +29,7 @@ from utils.utils import (
 from utils.misc import parse_duration, validate_email_format
 from utils.webhook import post_webhook
 from constants import ERROR_MESSAGES, WEBHOOK_MESSAGES
+from config import WEBUI_AUTH_TRUSTED_EMAIL_HEADER
 
 router = APIRouter()
 
@@ -79,6 +80,8 @@ async def update_profile(
 async def update_password(
     form_data: UpdatePasswordForm, session_user=Depends(get_current_user)
 ):
+    if WEBUI_AUTH_TRUSTED_EMAIL_HEADER:
+        raise HTTPException(400, detail=ERROR_MESSAGES.ACTION_PROHIBITED)
     if session_user:
         user = Auths.authenticate_user(session_user.email, form_data.password)
 
@@ -98,7 +101,22 @@ async def update_password(
 
 @router.post("/signin", response_model=SigninResponse)
 async def signin(request: Request, form_data: SigninForm):
-    user = Auths.authenticate_user(form_data.email.lower(), form_data.password)
+    if WEBUI_AUTH_TRUSTED_EMAIL_HEADER:
+        if WEBUI_AUTH_TRUSTED_EMAIL_HEADER not in request.headers:
+            raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_TRUSTED_HEADER)
+
+        trusted_email = request.headers[WEBUI_AUTH_TRUSTED_EMAIL_HEADER].lower()
+        if not Users.get_user_by_email(trusted_email.lower()):
+            await signup(
+                request,
+                SignupForm(
+                    email=trusted_email, password=str(uuid.uuid4()), name=trusted_email
+                ),
+            )
+        user = Auths.authenticate_user_by_trusted_header(trusted_email)
+    else:
+        user = Auths.authenticate_user(form_data.email.lower(), form_data.password)
+
     if user:
         token = create_token(
             data={"id": user.id},