Merge remote-tracking branch 'origin/master' into feat/sandbox

# Conflicts:
#	src/langbot/pkg/api/http/controller/groups/plugins.py
#	src/langbot/pkg/core/app.py
#	src/langbot/pkg/core/stages/build_app.py
#	src/langbot/templates/config.yaml
#	uv.lock
#	web/src/app/home/components/home-sidebar/HomeSidebar.tsx
#	web/src/app/home/components/home-sidebar/SidebarDataContext.tsx
#	web/src/app/home/layout.tsx
#	web/src/app/home/plugins/components/plugin-market/PluginMarketComponent.tsx
#	web/src/i18n/locales/en-US.ts
#	web/src/i18n/locales/es-ES.ts
#	web/src/i18n/locales/ja-JP.ts
#	web/src/i18n/locales/th-TH.ts
#	web/src/i18n/locales/vi-VN.ts
#	web/src/i18n/locales/zh-Hans.ts
#	web/src/i18n/locales/zh-Hant.ts
#	web/src/router.tsx

src/langbot/pkg/api/http/controller/groups/pipelines/embed.py

@@ -0,0 +1,384 @@
+"""Embed widget routes - serve embeddable chat widget for external websites.
+
+All user-facing URLs are keyed by **bot_uuid** (not pipeline_uuid) so that
+internal pipeline identifiers are never exposed to end-users.  Each handler
+resolves the bot_uuid to the owning ``web_page_bot`` RuntimeBot and extracts
+the bound pipeline_uuid for internal routing.
+"""
+
+import asyncio
+import datetime
+import json
+import logging
+import uuid
+import hmac
+import hashlib
+import time
+import re
+import httpx
+
+import quart
+
+from ... import group
+from ......utils import paths
+from ......platform.sources.websocket_manager import ws_connection_manager
+
+logger = logging.getLogger(__name__)
+
+# Cache the widget template content
+_widget_template_cache: str | None = None
+_logo_bytes_cache: bytes | None = None
+
+
+def _is_valid_uuid(s: str) -> bool:
+    return bool(re.match(r'^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$', s))
+
+
+def _get_widget_template() -> str:
+    """Load and cache the widget JS template."""
+    global _widget_template_cache
+    if _widget_template_cache is None:
+        template_path = paths.get_resource_path('templates/embed/widget.js')
+        with open(template_path, 'r', encoding='utf-8') as f:
+            _widget_template_cache = f.read()
+    return _widget_template_cache
+
+
+def _get_logo_bytes() -> bytes:
+    """Load and cache the logo image."""
+    global _logo_bytes_cache
+    if _logo_bytes_cache is None:
+        logo_path = paths.get_resource_path('templates/embed/logo.webp')
+        with open(logo_path, 'rb') as f:
+            _logo_bytes_cache = f.read()
+    return _logo_bytes_cache
+
+
+@group.group_class('embed', '/api/v1/embed')
+class EmbedRouterGroup(group.RouterGroup):
+    # -- helpers -------------------------------------------------------------
+
+    def _resolve_bot(self, bot_uuid: str):
+        """Resolve *bot_uuid* to ``(runtime_bot, pipeline_uuid)``.
+
+        Returns ``(None, None)`` when the bot does not exist, is not a
+        ``web_page_bot``, is disabled, or has no pipeline bound.
+        """
+        for bot in self.ap.platform_mgr.bots:
+            if (
+                bot.bot_entity.uuid == bot_uuid
+                and bot.bot_entity.adapter == 'web_page_bot'
+                and bot.bot_entity.enable
+                and bot.bot_entity.use_pipeline_uuid
+            ):
+                return bot, bot.bot_entity.use_pipeline_uuid
+        return None, None
+
+    def _get_bot_config(self, bot_uuid: str) -> dict:
+        for bot in self.ap.platform_mgr.bots:
+            if bot.bot_entity.uuid == bot_uuid and bot.bot_entity.adapter == 'web_page_bot':
+                return bot.bot_entity.adapter_config
+        return {}
+
+    async def _verify_session_token(self, request, bot_uuid: str) -> bool:
+        config = self._get_bot_config(bot_uuid)
+        secret = config.get('turnstile_secret_key', '')
+        if not secret:
+            return True
+        auth_header = request.headers.get('Authorization', '')
+        if not auth_header.startswith('Bearer '):
+            return False
+        token = auth_header[7:]
+        try:
+            ts_str, mac = token.split('.', 1)
+            ts = float(ts_str)
+            if time.time() - ts > 86400:
+                return False
+            expected_mac = hmac.new(secret.encode(), f'{ts_str}'.encode(), hashlib.sha256).hexdigest()
+            return hmac.compare_digest(mac, expected_mac)
+        except Exception:
+            return False
+
+    # -- routes --------------------------------------------------------------
+
+    async def initialize(self) -> None:
+        @self.route('/<bot_uuid>/turnstile/verify', methods=['POST'], auth_type=group.AuthType.NONE)
+        async def verify_turnstile(bot_uuid: str) -> str:
+            if not _is_valid_uuid(bot_uuid):
+                return self.http_status(400, -1, 'Invalid bot_uuid format')
+            runtime_bot, pipeline_uuid = self._resolve_bot(bot_uuid)
+            if runtime_bot is None:
+                return self.http_status(404, -1, 'Bot not found or not available')
+            try:
+                data = await quart.request.get_json()
+                token = data.get('token')
+                if not token:
+                    return self.http_status(400, -1, 'Token is required')
+
+                config = self._get_bot_config(bot_uuid)
+                secret = config.get('turnstile_secret_key', '')
+                if not secret:
+                    ts = time.time()
+                    return self.success(data={'token': f'{ts}.dummy'})
+
+                async with httpx.AsyncClient() as client:
+                    resp = await client.post(
+                        'https://challenges.cloudflare.com/turnstile/v0/siteverify',
+                        data={'secret': secret, 'response': token},
+                    )
+                    result = resp.json()
+
+                if not result.get('success'):
+                    return self.http_status(403, -1, 'Turnstile verification failed')
+
+                ts = time.time()
+                mac = hmac.new(secret.encode(), f'{ts}'.encode(), hashlib.sha256).hexdigest()
+                session_token = f'{ts}.{mac}'
+
+                return self.success(data={'token': session_token})
+
+            except Exception as e:
+                logger.error(f'Turnstile verify failed: {e}', exc_info=True)
+                return self.http_status(500, -1, 'Internal server error')
+
+        @self.route('/<bot_uuid>/widget.js', methods=['GET'], auth_type=group.AuthType.NONE)
+        async def serve_widget(bot_uuid: str) -> quart.Response:
+            """Serve the embed widget JavaScript with injected configuration."""
+            if not _is_valid_uuid(bot_uuid):
+                return self.http_status(400, -1, 'Invalid bot_uuid format')
+            runtime_bot, pipeline_uuid = self._resolve_bot(bot_uuid)
+            if runtime_bot is None:
+                return quart.Response(
+                    '// Bot not found or not available', status=404, content_type='application/javascript'
+                )
+            try:
+                template = _get_widget_template()
+            except FileNotFoundError:
+                return quart.Response('// Widget template not found', status=404, content_type='application/javascript')
+
+            base_url = quart.request.host_url.rstrip('/')
+            webhook_prefix = self.ap.instance_config.data.get('api', {}).get('webhook_prefix', '')
+            if webhook_prefix:
+                base_url = webhook_prefix.rstrip('/')
+
+            if not re.match(r'^https?://[a-zA-Z0-9._:/-]+$', base_url):
+                base_url = quart.request.host_url.rstrip('/')
+
+            config = self._get_bot_config(bot_uuid)
+            site_key = config.get('turnstile_site_key', '')
+            locale = config.get('language', 'en_US') or 'en_US'
+            bubble_icon = config.get('bubble_icon', 'logo') or 'logo'
+            widget_js = template.replace('__LANGBOT_TURNSTILE_SITE_KEY__', site_key)
+            widget_js = widget_js.replace('__LANGBOT_BOT_UUID__', bot_uuid)
+            widget_js = widget_js.replace('__LANGBOT_BASE_URL__', base_url)
+            widget_js = widget_js.replace('__LANGBOT_LOCALE__', locale)
+            widget_js = widget_js.replace('__LANGBOT_BUBBLE_ICON__', bubble_icon)
+
+            response = quart.Response(widget_js, content_type='application/javascript; charset=utf-8')
+            response.headers['Cache-Control'] = 'public, max-age=300'
+            return response
+
+        @self.route('/logo', methods=['GET'], auth_type=group.AuthType.NONE)
+        async def serve_logo() -> quart.Response:
+            """Serve the LangBot logo for the embed widget."""
+            try:
+                logo_data = _get_logo_bytes()
+            except FileNotFoundError:
+                return quart.Response('', status=404)
+
+            response = quart.Response(logo_data, content_type='image/webp')
+            response.headers['Cache-Control'] = 'public, max-age=86400'
+            return response
+
+        @self.route('/<bot_uuid>/messages/<session_type>', methods=['GET'], auth_type=group.AuthType.NONE)
+        async def get_embed_messages(bot_uuid: str, session_type: str) -> str:
+            if not _is_valid_uuid(bot_uuid):
+                return self.http_status(400, -1, 'Invalid bot_uuid format')
+            runtime_bot, pipeline_uuid = self._resolve_bot(bot_uuid)
+            if runtime_bot is None:
+                return self.http_status(404, -1, 'Bot not found or not available')
+            if not await self._verify_session_token(quart.request, bot_uuid):
+                return self.http_status(403, -1, 'Unauthorized or session expired')
+            try:
+                if session_type not in ['person', 'group']:
+                    return self.http_status(400, -1, 'session_type must be person or group')
+
+                websocket_adapter = self.ap.platform_mgr.websocket_proxy_bot.adapter
+                if not websocket_adapter:
+                    return self.http_status(404, -1, 'WebSocket adapter not found')
+
+                messages = websocket_adapter.get_websocket_messages(pipeline_uuid, session_type)
+                return self.success(data={'messages': messages})
+
+            except Exception as e:
+                logger.error(f'Failed to get embed messages: {e}', exc_info=True)
+                return self.http_status(500, -1, 'Internal server error')
+
+        @self.route('/<bot_uuid>/reset/<session_type>', methods=['POST'], auth_type=group.AuthType.NONE)
+        async def reset_embed_session(bot_uuid: str, session_type: str) -> str:
+            if not _is_valid_uuid(bot_uuid):
+                return self.http_status(400, -1, 'Invalid bot_uuid format')
+            runtime_bot, pipeline_uuid = self._resolve_bot(bot_uuid)
+            if runtime_bot is None:
+                return self.http_status(404, -1, 'Bot not found or not available')
+            if not await self._verify_session_token(quart.request, bot_uuid):
+                return self.http_status(403, -1, 'Unauthorized or session expired')
+            try:
+                if session_type not in ['person', 'group']:
+                    return self.http_status(400, -1, 'session_type must be person or group')
+
+                websocket_adapter = self.ap.platform_mgr.websocket_proxy_bot.adapter
+                if not websocket_adapter:
+                    return self.http_status(404, -1, 'WebSocket adapter not found')
+
+                websocket_adapter.reset_session(pipeline_uuid, session_type)
+                return self.success(data={'message': 'Session reset successfully'})
+
+            except Exception as e:
+                logger.error(f'Failed to reset embed session: {e}', exc_info=True)
+                return self.http_status(500, -1, 'Internal server error')
+
+        @self.route('/<bot_uuid>/feedback', methods=['POST'], auth_type=group.AuthType.NONE)
+        async def submit_feedback(bot_uuid: str) -> str:
+            if not _is_valid_uuid(bot_uuid):
+                return self.http_status(400, -1, 'Invalid bot_uuid format')
+            runtime_bot, pipeline_uuid = self._resolve_bot(bot_uuid)
+            if runtime_bot is None:
+                return self.http_status(404, -1, 'Bot not found or not available')
+            if not await self._verify_session_token(quart.request, bot_uuid):
+                return self.http_status(403, -1, 'Unauthorized or session expired')
+            try:
+                data = await quart.request.get_json()
+                message_id = data.get('message_id', '')
+                feedback_type = data.get('feedback_type')
+
+                if feedback_type not in (1, 2, 3):
+                    return self.http_status(400, -1, 'feedback_type must be 1 (like), 2 (dislike), or 3 (cancel)')
+
+                feedback_id = f'embed_{uuid.uuid4().hex[:12]}'
+
+                await self.ap.monitoring_service.record_feedback(
+                    feedback_id=feedback_id,
+                    feedback_type=feedback_type,
+                    bot_id=runtime_bot.bot_entity.uuid,
+                    bot_name=runtime_bot.bot_entity.name or bot_uuid,
+                    pipeline_id=pipeline_uuid,
+                    message_id=str(message_id),
+                    platform='web_page_bot',
+                )
+
+                return self.success(data={'feedback_id': feedback_id})
+
+            except Exception as e:
+                logger.error(f'Failed to record feedback: {e}', exc_info=True)
+                return self.http_status(500, -1, 'Internal server error')
+
+        # -- Embed WebSocket endpoint ----------------------------------------
+
+        @self.quart_app.websocket(self.path + '/<bot_uuid>/ws/connect')
+        async def embed_websocket_connect(bot_uuid: str):
+            """WebSocket connection for embed widget, keyed by bot_uuid."""
+            if not _is_valid_uuid(bot_uuid):
+                await quart.websocket.send(json.dumps({'type': 'error', 'message': 'Invalid bot_uuid format'}))
+                return
+
+            runtime_bot, pipeline_uuid = self._resolve_bot(bot_uuid)
+            if runtime_bot is None:
+                await quart.websocket.send(json.dumps({'type': 'error', 'message': 'Bot not found or not available'}))
+                return
+
+            session_type = quart.websocket.args.get('session_type', 'person')
+            if session_type not in ['person', 'group']:
+                await quart.websocket.send(
+                    json.dumps({'type': 'error', 'message': 'session_type must be person or group'})
+                )
+                return
+
+            websocket_adapter = self.ap.platform_mgr.websocket_proxy_bot.adapter
+            if not websocket_adapter:
+                await quart.websocket.send(json.dumps({'type': 'error', 'message': 'WebSocket adapter not found'}))
+                return
+
+            try:
+                connection = await ws_connection_manager.add_connection(
+                    websocket=quart.websocket._get_current_object(),
+                    pipeline_uuid=pipeline_uuid,
+                    session_type=session_type,
+                    metadata={'user_agent': quart.websocket.headers.get('User-Agent', '')},
+                )
+
+                await quart.websocket.send(
+                    json.dumps(
+                        {
+                            'type': 'connected',
+                            'connection_id': connection.connection_id,
+                            'bot_uuid': bot_uuid,
+                            'session_type': session_type,
+                            'timestamp': connection.created_at.isoformat(),
+                        }
+                    )
+                )
+
+                logger.debug(
+                    f'Embed WebSocket connected: {connection.connection_id} '
+                    f'(bot={bot_uuid}, pipeline={pipeline_uuid}, session_type={session_type})'
+                )
+
+                receive_task = asyncio.create_task(self._handle_receive(connection, websocket_adapter, runtime_bot))
+                send_task = asyncio.create_task(self._handle_send(connection))
+
+                try:
+                    await asyncio.gather(receive_task, send_task)
+                except Exception as e:
+                    logger.error(f'Embed WebSocket task error: {e}')
+                finally:
+                    await ws_connection_manager.remove_connection(connection.connection_id)
+
+            except Exception as e:
+                logger.error(f'Embed WebSocket connection error: {e}', exc_info=True)
+                try:
+                    await quart.websocket.send(json.dumps({'type': 'error', 'message': 'Internal server error'}))
+                except Exception:
+                    pass
+
+    # -- WebSocket receive/send helpers --------------------------------------
+
+    async def _handle_receive(self, connection, websocket_adapter, owner_bot):
+        try:
+            while connection.is_active:
+                message = await quart.websocket.receive()
+                await ws_connection_manager.update_activity(connection.connection_id)
+
+                try:
+                    data = json.loads(message)
+                    message_type = data.get('type', 'message')
+
+                    if message_type == 'ping':
+                        await connection.send_queue.put(
+                            {'type': 'pong', 'timestamp': datetime.datetime.now().isoformat()}
+                        )
+                    elif message_type == 'message':
+                        await websocket_adapter.handle_websocket_message(connection, data, owner_bot=owner_bot)
+                    elif message_type == 'disconnect':
+                        break
+
+                except json.JSONDecodeError:
+                    await connection.send_queue.put({'type': 'error', 'message': 'Invalid JSON format'})
+
+        except Exception as e:
+            logger.error(f'Embed receive error: {e}', exc_info=True)
+        finally:
+            connection.is_active = False
+
+    async def _handle_send(self, connection):
+        try:
+            while connection.is_active:
+                try:
+                    message = await asyncio.wait_for(connection.send_queue.get(), timeout=1.0)
+                    await quart.websocket.send(json.dumps(message))
+                except asyncio.TimeoutError:
+                    continue
+        except Exception as e:
+            logger.error(f'Embed send error: {e}', exc_info=True)
+        finally:
+            connection.is_active = False

src/langbot/pkg/api/http/controller/groups/pipelines/websocket_chat.py

@@ -43,6 +43,9 @@ class WebSocketChatRouterGroup(group.RouterGroup):
                     await quart.websocket.send(json.dumps({'type': 'error', 'message': 'WebSocket adapter not found'}))
                     return
 
+                # Find the owning bot for this pipeline (e.g. a web_page_bot)
+                owner_bot = self._find_owner_bot(pipeline_uuid)
+
                 # 注册连接
                 connection = await ws_connection_manager.add_connection(
                     websocket=quart.websocket._get_current_object(),
@@ -70,7 +73,7 @@ class WebSocketChatRouterGroup(group.RouterGroup):
                 )
 
                 # 创建接收和发送任务
-                receive_task = asyncio.create_task(self._handle_receive(connection, websocket_adapter))
+                receive_task = asyncio.create_task(self._handle_receive(connection, websocket_adapter, owner_bot))
                 send_task = asyncio.create_task(self._handle_send(connection))
 
                 # 等待任务完成
@@ -178,7 +181,14 @@ class WebSocketChatRouterGroup(group.RouterGroup):
             except Exception as e:
                 return self.http_status(500, -1, f'Internal server error: {str(e)}')
 
-    async def _handle_receive(self, connection, websocket_adapter):
+    def _find_owner_bot(self, pipeline_uuid: str):
+        """Find a user-created bot (e.g. web_page_bot) that owns this pipeline."""
+        for bot in self.ap.platform_mgr.bots:
+            if bot.bot_entity.adapter == 'web_page_bot' and bot.bot_entity.use_pipeline_uuid == pipeline_uuid:
+                return bot
+        return None
+
+    async def _handle_receive(self, connection, websocket_adapter, owner_bot=None):
         """处理接收消息的任务"""
         try:
             while connection.is_active:
@@ -203,7 +213,7 @@ class WebSocketChatRouterGroup(group.RouterGroup):
                         logger.debug(f'收到消息: {data} from {connection.connection_id}')
 
                         # 处理消息（不等待响应，响应会通过broadcast异步发送）
-                        await websocket_adapter.handle_websocket_message(connection, data)
+                        await websocket_adapter.handle_websocket_message(connection, data, owner_bot=owner_bot)
 
                     elif message_type == 'disconnect':
                         # 客户端主动断开

src/langbot/pkg/api/http/controller/groups/plugins.py

@@ -7,11 +7,38 @@ import httpx
 import uuid
 import os
 from urllib.parse import urlparse
+import posixpath
 
 from .....core import taskmgr
 from .. import group
 from langbot_plugin.runtime.plugin.mgr import PluginInstallSource
 
+# Resolve the built-in page SDK JS from the langbot_plugin package
+_PAGE_SDK_PATH = None
+try:
+    import langbot_plugin.assets as _assets_pkg
+
+    _candidate = os.path.join(os.path.dirname(_assets_pkg.__file__), 'langbot-page-sdk.js')
+    if os.path.exists(_candidate):
+        _PAGE_SDK_PATH = _candidate
+except Exception:
+    pass
+
+
+def _normalize_plugin_asset_path(filepath: str) -> str | None:
+    filepath = filepath.replace('\\', '/')
+    if filepath.startswith('/'):
+        return None
+
+    normalized = posixpath.normpath(filepath)
+    if normalized == '.' or normalized.startswith('../') or normalized == '..':
+        return None
+
+    if normalized.startswith('components/pages/'):
+        return normalized
+
+    return f'assets/{normalized}'
+
 
 @group.group_class('plugins', '/api/v1/plugins')
 class PluginsRouterGroup(group.RouterGroup):
@@ -65,6 +92,15 @@ class PluginsRouterGroup(group.RouterGroup):
         return None
 
     async def initialize(self) -> None:
+        @self.route('/_sdk/page-sdk.js', methods=['GET'], auth_type=group.AuthType.NONE)
+        async def _() -> quart.Response:
+            """Serve the built-in LangBot page SDK JavaScript."""
+            if _PAGE_SDK_PATH and os.path.exists(_PAGE_SDK_PATH):
+                with open(_PAGE_SDK_PATH, 'r') as f:
+                    content = f.read()
+                return quart.Response(content, mimetype='application/javascript')
+            return quart.Response('// SDK not found', status=404, mimetype='application/javascript')
+
         @self.route('', methods=['GET'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
         async def _() -> str:
             plugins = await self.ap.plugin_connector.list_plugins()
@@ -173,15 +209,62 @@ class PluginsRouterGroup(group.RouterGroup):
             return quart.Response(icon_data, mimetype=mime_type)
 
         @self.route(
-            '/<author>/<plugin_name>/assets/<filepath>',
+            '/<author>/<plugin_name>/assets/<path:filepath>',
             methods=['GET'],
             auth_type=group.AuthType.NONE,
         )
         async def _(author: str, plugin_name: str, filepath: str) -> quart.Response:
-            asset_data = await self.ap.plugin_connector.get_plugin_assets(author, plugin_name, filepath)
+            asset_path = _normalize_plugin_asset_path(filepath)
+            if asset_path is None:
+                return quart.Response('Asset not found', status=404)
+
+            asset_data = await self.ap.plugin_connector.get_plugin_assets(author, plugin_name, asset_path)
+            if not asset_data.get('asset_base64'):
+                return quart.Response('Asset not found', status=404)
             asset_bytes = base64.b64decode(asset_data['asset_base64'])
             mime_type = asset_data['mime_type']
-            return quart.Response(asset_bytes, mimetype=mime_type)
+            resp = quart.Response(asset_bytes, mimetype=mime_type)
+            # CSP for HTML pages served to sandboxed iframes (opaque origin).
+            # 'self' doesn't work in sandboxed iframes — use actual server origin.
+            if mime_type and mime_type.startswith('text/html'):
+                origin = f'{quart.request.scheme}://{quart.request.host}'
+                resp.headers['Content-Security-Policy'] = (
+                    f'default-src {origin}; '
+                    f"script-src {origin} 'unsafe-inline'; "
+                    f"style-src {origin} 'unsafe-inline'; "
+                    f'img-src {origin} data:; '
+                    f'connect-src {origin}; '
+                    "frame-src 'none'; "
+                    "object-src 'none'"
+                )
+            return resp
+
+        @self.route(
+            '/<author>/<plugin_name>/page-api',
+            methods=['POST'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+        )
+        async def _(author: str, plugin_name: str) -> str:
+            """Forward a page API request to the plugin."""
+            data = await quart.request.json
+            if not isinstance(data, dict):
+                return self.http_status(400, -1, 'invalid request body')
+
+            page_id = data.get('page_id', '')
+            endpoint = data.get('endpoint', '')
+            method = data.get('method', 'POST')
+            body = data.get('body')
+            if not isinstance(page_id, str) or not isinstance(endpoint, str) or not isinstance(method, str):
+                return self.http_status(400, -1, 'invalid page api request')
+            if not endpoint.startswith('/') or '..' in endpoint:
+                return self.http_status(400, -1, 'invalid endpoint')
+
+            result = await self.ap.plugin_connector.handle_page_api(
+                author, plugin_name, page_id, endpoint, method.upper(), body
+            )
+            if result.get('error'):
+                return self.http_status(400, -1, result['error'])
+            return self.success(data=result.get('data'))
 
         @self.route('/github/releases', methods=['POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
         async def _() -> str:

src/langbot/pkg/api/http/controller/groups/system.py

@@ -136,6 +136,10 @@ class SystemRouterGroup(group.RouterGroup):
 
             return self.success(data=task.to_dict())
 
+        @self.route('/storage-analysis', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
+        async def _() -> str:
+            return self.success(data=await self.ap.maintenance_service.get_storage_analysis())
+
         @self.route('/debug/exec', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
         async def _() -> str:
             if not constants.debug_mode: