Feat：admin api (#10642)

### What problem does this PR solve?

Support frontend auth.

### Type of change

- [x] New Feature (non-breaking change which adds functionality)

admin/client/admin_client.py

@@ -23,7 +23,6 @@ from Cryptodome.Cipher import PKCS1_v1_5 as Cipher_pkcs1_v1_5
 from typing import Dict, List, Any
 from lark import Lark, Transformer, Tree, Token
 import requests
-from requests.auth import HTTPBasicAuth
 
 GRAMMAR = r"""
 start: command
@@ -205,6 +204,8 @@ class AdminCLI(Cmd):
         self.is_interactive = False
         self.admin_account = "admin@ragflow.io"
         self.admin_password: str = "admin"
+        self.session = requests.Session()
+        self.access_token: str = ""
         self.host: str = ""
         self.port: int = 0
 
@@ -262,7 +263,7 @@ class AdminCLI(Cmd):
         self.host = conn_info['host']
         self.port = conn_info['port']
         print(f"Attempt to access ip: {self.host}, port: {self.port}")
-        url = f'http://{self.host}:{self.port}/api/v1/admin/auth'
+        url = f"http://{self.host}:{self.port}/api/v1/admin/login"
 
         try_count = 0
         while True:
@@ -272,12 +273,17 @@ class AdminCLI(Cmd):
 
             admin_passwd = input(f"password for {self.admin_account}: ").strip()
             try:
-                self.admin_password = encode_to_base64(admin_passwd)
+                self.admin_password = encrypt(admin_passwd)
-                response = requests.get(url, auth=HTTPBasicAuth(self.admin_account, self.admin_password))
+                response = self.session.post(url, json={'email': self.admin_account, 'password': self.admin_password})
                 if response.status_code == 200:
                     res_json = response.json()
                     error_code = res_json.get('code', -1)
                     if error_code == 0:
+                        self.session.headers.update({
+                            'Content-Type': 'application/json',
+                            'Authorization': response.headers['Authorization'],
+                            'User-Agent': 'RAGFlow-CLI/0.21.0'
+                        })
                         print("Authentication successful.")
                         return True
                     else:
@@ -286,7 +292,8 @@ class AdminCLI(Cmd):
                         continue
                 else:
                     print(f"Bad response，status: {response.status_code}, try again")
-            except Exception:
+            except Exception as e:
+                print(str(e))
                 print(f"Can't access {self.host}, port: {self.port}")
 
     def _print_table_simple(self, data):
@@ -443,7 +450,7 @@ class AdminCLI(Cmd):
         print("Listing all services")
 
         url = f'http://{self.host}:{self.port}/api/v1/admin/services'
-        response = requests.get(url, auth=HTTPBasicAuth(self.admin_account, self.admin_password))
+        response = self.session.get(url)
         res_json = response.json()
         if response.status_code == 200:
             self._print_table_simple(res_json['data'])
@@ -455,7 +462,7 @@ class AdminCLI(Cmd):
         print(f"Showing service: {service_id}")
 
         url = f'http://{self.host}:{self.port}/api/v1/admin/services/{service_id}'
-        response = requests.get(url, auth=HTTPBasicAuth(self.admin_account, self.admin_password))
+        response = self.session.get(url)
         res_json = response.json()
         if response.status_code == 200:
             res_data = res_json['data']
@@ -486,7 +493,7 @@ class AdminCLI(Cmd):
         print("Listing all users")
 
         url = f'http://{self.host}:{self.port}/api/v1/admin/users'
-        response = requests.get(url, auth=HTTPBasicAuth(self.admin_account, self.admin_password))
+        response = self.session.get(url)
         res_json = response.json()
         if response.status_code == 200:
             self._print_table_simple(res_json['data'])
@@ -498,7 +505,7 @@ class AdminCLI(Cmd):
         username: str = username_tree.children[0].strip("'\"")
         print(f"Showing user: {username}")
         url = f'http://{self.host}:{self.port}/api/v1/admin/users/{username}'
-        response = requests.get(url, auth=HTTPBasicAuth(self.admin_account, self.admin_password))
+        response = self.session.get(url)
         res_json = response.json()
         if response.status_code == 200:
             self._print_table_simple(res_json['data'])
@@ -510,7 +517,7 @@ class AdminCLI(Cmd):
         username: str = username_tree.children[0].strip("'\"")
         print(f"Drop user: {username}")
         url = f'http://{self.host}:{self.port}/api/v1/admin/users/{username}'
-        response = requests.delete(url, auth=HTTPBasicAuth(self.admin_account, self.admin_password))
+        response = self.session.delete(url)
         res_json = response.json()
         if response.status_code == 200:
             print(res_json["message"])
@@ -524,8 +531,7 @@ class AdminCLI(Cmd):
         password: str = password_tree.children[0].strip("'\"")
         print(f"Alter user: {username}, password: {password}")
         url = f'http://{self.host}:{self.port}/api/v1/admin/users/{username}/password'
-        response = requests.put(url, auth=HTTPBasicAuth(self.admin_account, self.admin_password),
+        response = self.session.put(url, json={'new_password': encrypt(password)})
-                                json={'new_password': encrypt(password)})
         res_json = response.json()
         if response.status_code == 200:
             print(res_json["message"])
@@ -540,9 +546,8 @@ class AdminCLI(Cmd):
         role: str = command['role']
         print(f"Create user: {username}, password: {password}, role: {role}")
         url = f'http://{self.host}:{self.port}/api/v1/admin/users'
-        response = requests.post(
+        response = self.session.post(
             url,
-            auth=HTTPBasicAuth(self.admin_account, self.admin_password),
             json={'username': username, 'password': encrypt(password), 'role': role}
         )
         res_json = response.json()
@@ -559,8 +564,7 @@ class AdminCLI(Cmd):
         if activate_status.lower() in ['on', 'off']:
             print(f"Alter user {username} activate status, turn {activate_status.lower()}.")
             url = f'http://{self.host}:{self.port}/api/v1/admin/users/{username}/activate'
-            response = requests.put(url, auth=HTTPBasicAuth(self.admin_account, self.admin_password),
+            response = self.session.put(url, json={'activate_status': activate_status})
-                                    json={'activate_status': activate_status})
             res_json = response.json()
             if response.status_code == 200:
                 print(res_json["message"])
@@ -574,7 +578,7 @@ class AdminCLI(Cmd):
         username: str = username_tree.children[0].strip("'\"")
         print(f"Listing all datasets of user: {username}")
         url = f'http://{self.host}:{self.port}/api/v1/admin/users/{username}/datasets'
-        response = requests.get(url, auth=HTTPBasicAuth(self.admin_account, self.admin_password))
+        response = self.session.get(url)
         res_json = response.json()
         if response.status_code == 200:
             self._print_table_simple(res_json['data'])
@@ -586,7 +590,7 @@ class AdminCLI(Cmd):
         username: str = username_tree.children[0].strip("'\"")
         print(f"Listing all agents of user: {username}")
         url = f'http://{self.host}:{self.port}/api/v1/admin/users/{username}/agents'
-        response = requests.get(url, auth=HTTPBasicAuth(self.admin_account, self.admin_password))
+        response = self.session.get(url)
         res_json = response.json()
         if response.status_code == 200:
             self._print_table_simple(res_json['data'])

admin/server/admin_server.py

@@ -27,6 +27,9 @@ from api.utils.log_utils import init_root_logger
 from api.constants import SERVICE_CONF
 from api import settings
 from config import load_configurations, SERVICE_CONFIGS
+from auth import init_default_admin, setup_auth
+from flask_session import Session
+from flask_login import LoginManager
 
 stop_event = threading.Event()
 
@@ -42,7 +45,17 @@ if __name__ == '__main__':
 
     app = Flask(__name__)
     app.register_blueprint(admin_bp)
+    app.config["SESSION_PERMANENT"] = False
+    app.config["SESSION_TYPE"] = "filesystem"
+    app.config["MAX_CONTENT_LENGTH"] = int(
+        os.environ.get("MAX_CONTENT_LENGTH", 1024 * 1024 * 1024)
+    )
+    Session(app)
+    login_manager = LoginManager()
+    login_manager.init_app(app)
     settings.init_settings()
+    setup_auth(login_manager)
+    init_default_admin()
     SERVICE_CONFIGS.configs = load_configurations(SERVICE_CONF)
 
     try:

admin/server/auth.py

@@ -18,11 +18,122 @@
 import logging
 import uuid
 from functools import wraps
+from datetime import datetime
 from flask import request, jsonify
+from flask_login import current_user, login_user
+from itsdangerous.url_safe import URLSafeTimedSerializer as Serializer
 
-from api.common.exceptions import AdminException
+from api import settings
+from api.common.exceptions import AdminException, UserNotFoundError
 from api.db.init_data import encode_to_base64
 from api.db.services import UserService
+from api.db import ActiveEnum, StatusEnum
+from api.utils.crypt import decrypt
+from api.utils import (
+    current_timestamp,
+    datetime_format,
+    get_uuid,
+)
+from api.utils.api_utils import (
+    construct_response,
+)
+
+def setup_auth(login_manager):
+
+    @login_manager.request_loader
+    def load_user(web_request):
+        jwt = Serializer(secret_key=settings.SECRET_KEY)
+        authorization = web_request.headers.get("Authorization")
+        if authorization:
+            try:
+                access_token = str(jwt.loads(authorization))
+
+                if not access_token or not access_token.strip():
+                    logging.warning("Authentication attempt with empty access token")
+                    return None
+
+                # Access tokens should be UUIDs (32 hex characters)
+                if len(access_token.strip()) < 32:
+                    logging.warning(f"Authentication attempt with invalid token format: {len(access_token)} chars")
+                    return None
+
+                user = UserService.query(
+                    access_token=access_token, status=StatusEnum.VALID.value
+                )
+                if user:
+                    if not user[0].access_token or not user[0].access_token.strip():
+                        logging.warning(f"User {user[0].email} has empty access_token in database")
+                        return None
+                    return user[0]
+                else:
+                    return None
+            except Exception as e:
+                logging.warning(f"load_user got exception {e}")
+                return None
+        else:
+            return None
+
+
+def init_default_admin():
+    # Verify that at least one active admin user exists. If not, create a default one.
+    users = UserService.query(is_superuser=True)
+    if not users:
+        default_admin = {
+            "id": uuid.uuid1().hex,
+            "password": encode_to_base64("admin"),
+            "nickname": "admin",
+            "is_superuser": True,
+            "email": "admin@ragflow.io",
+            "creator": "system",
+            "status": "1",
+        }
+        if not UserService.save(**default_admin):
+            raise AdminException("Can't init admin.", 500)
+    elif not any([u.is_active == ActiveEnum.ACTIVE.value for u in users]):
+        raise AdminException("No active admin. Please update 'is_active' in db manually.", 500)
+
+
+def check_admin_auth(func):
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        user = UserService.filter_by_id(current_user.id)
+        if not user:
+            raise UserNotFoundError(current_user.email)
+        if not user.is_superuser:
+            raise AdminException("Not admin", 403)
+        if user.is_active == ActiveEnum.INACTIVE.value:
+            raise AdminException(f"User {current_user.email} inactive", 403)
+
+        return func(*args, **kwargs)
+
+    return wrapper
+
+
+def login_admin(email: str, password: str):
+    """
+    :param email: admin email
+    :param password: string before decrypt
+    """
+    users = UserService.query(email=email)
+    if not users:
+        raise UserNotFoundError(email)
+    psw = decrypt(password)
+    user = UserService.query_user(email, psw)
+    if not user:
+        raise AdminException("Email and password do not match!")
+    if not user.is_superuser:
+        raise AdminException("Not admin", 403)
+    if user.is_active == ActiveEnum.INACTIVE.value:
+        raise AdminException(f"User {email} inactive", 403)
+
+    resp = user.to_json()
+    user.access_token = get_uuid()
+    login_user(user)
+    user.update_time = (current_timestamp(),)
+    user.update_date = (datetime_format(datetime.now()),)
+    user.save()
+    msg = "Welcome back!"
+    return construct_response(data=resp, auth=user.get_id(), message=msg)
 
 
 def check_admin(username: str, password: str):
@@ -61,7 +172,6 @@ def login_verify(f):
 
         username = auth.parameters['username']
         password = auth.parameters['password']
-        # TODO: to check the username and password from DB
         if check_admin(username, password) is False:
             return jsonify({
                 "code": 403,

admin/server/routes.py

@@ -14,10 +14,12 @@
 #  limitations under the License.
 #
 
+import secrets
 
 from flask import Blueprint, request
+from flask_login import current_user, logout_user, login_required
 
-from auth import login_verify
+from auth import login_verify, login_admin, check_admin_auth
 from responses import success_response, error_response
 from services import UserMgr, ServiceMgr, UserServiceMgr
 from api.common.exceptions import AdminException
@@ -25,6 +27,24 @@ from api.common.exceptions import AdminException
 admin_bp = Blueprint('admin', __name__, url_prefix='/api/v1/admin')
 
 
+@admin_bp.route('/login', methods=['POST'])
+def login():
+    if not request.json:
+        return error_response('Authorize admin failed.' ,400)
+    email = request.json.get("email", "")
+    password = request.json.get("password", "")
+    return login_admin(email, password)
+
+
+@admin_bp.route('/logout', methods=['GET'])
+@login_required
+def logout():
+    current_user.access_token = f"INVALID_{secrets.token_hex(16)}"
+    current_user.save()
+    logout_user()
+    return success_response(True)
+
+
 @admin_bp.route('/auth', methods=['GET'])
 @login_verify
 def auth_admin():
@@ -35,7 +55,8 @@ def auth_admin():
 
 
 @admin_bp.route('/users', methods=['GET'])
-@login_verify
+@login_required
+@check_admin_auth
 def list_users():
     try:
         users = UserMgr.get_all_users()
@@ -45,7 +66,8 @@ def list_users():
 
 
 @admin_bp.route('/users', methods=['POST'])
-@login_verify
+@login_required
+@check_admin_auth
 def create_user():
     try:
         data = request.get_json()
@@ -71,7 +93,8 @@ def create_user():
 
 
 @admin_bp.route('/users/<username>', methods=['DELETE'])
-@login_verify
+@login_required
+@check_admin_auth
 def delete_user(username):
     try:
         res = UserMgr.delete_user(username)
@@ -87,7 +110,8 @@ def delete_user(username):
 
 
 @admin_bp.route('/users/<username>/password', methods=['PUT'])
-@login_verify
+@login_required
+@check_admin_auth
 def change_password(username):
     try:
         data = request.get_json()
@@ -105,7 +129,8 @@ def change_password(username):
 
 
 @admin_bp.route('/users/<username>/activate', methods=['PUT'])
-@login_verify
+@login_required
+@check_admin_auth
 def alter_user_activate_status(username):
     try:
         data = request.get_json()
@@ -121,7 +146,8 @@ def alter_user_activate_status(username):
 
 
 @admin_bp.route('/users/<username>', methods=['GET'])
-@login_verify
+@login_required
+@check_admin_auth
 def get_user_details(username):
     try:
         user_details = UserMgr.get_user_details(username)
@@ -134,7 +160,8 @@ def get_user_details(username):
 
 
 @admin_bp.route('/users/<username>/datasets', methods=['GET'])
-@login_verify
+@login_required
+@check_admin_auth
 def get_user_datasets(username):
     try:
         datasets_list = UserServiceMgr.get_user_datasets(username)
@@ -147,7 +174,8 @@ def get_user_datasets(username):
 
 
 @admin_bp.route('/users/<username>/agents', methods=['GET'])
-@login_verify
+@login_required
+@check_admin_auth
 def get_user_agents(username):
     try:
         agents_list = UserServiceMgr.get_user_agents(username)
@@ -160,7 +188,8 @@ def get_user_agents(username):
 
 
 @admin_bp.route('/services', methods=['GET'])
-@login_verify
+@login_required
+@check_admin_auth
 def get_services():
     try:
         services = ServiceMgr.get_all_services()
@@ -170,7 +199,8 @@ def get_services():
 
 
 @admin_bp.route('/service_types/<service_type>', methods=['GET'])
-@login_verify
+@login_required
+@check_admin_auth
 def get_services_by_type(service_type_str):
     try:
         services = ServiceMgr.get_services_by_type(service_type_str)
@@ -180,7 +210,8 @@ def get_services_by_type(service_type_str):
 
 
 @admin_bp.route('/services/<service_id>', methods=['GET'])
-@login_verify
+@login_required
+@check_admin_auth
 def get_service(service_id):
     try:
         services = ServiceMgr.get_service_details(service_id)
@@ -190,7 +221,8 @@ def get_service(service_id):
 
 
 @admin_bp.route('/services/<service_id>', methods=['DELETE'])
-@login_verify
+@login_required
+@check_admin_auth
 def shutdown_service(service_id):
     try:
         services = ServiceMgr.shutdown_service(service_id)
@@ -200,7 +232,8 @@ def shutdown_service(service_id):
 
 
 @admin_bp.route('/services/<service_id>', methods=['PUT'])
-@login_verify
+@login_required
+@check_admin_auth
 def restart_service(service_id):
     try:
         services = ServiceMgr.restart_service(service_id)

api/common/exceptions.py

@@ -36,3 +36,8 @@ class UserAlreadyExistsError(AdminException):
 class CannotDeleteAdminError(AdminException):
     def __init__(self):
         super().__init__("Cannot delete admin account", 403)
+
+
+class NotAdminError(AdminException):
+    def __init__(self, username):
+        super().__init__(f"User '{username}' is not admin", 403)