From 63bd5ff8a39e0036a1ccc5985a9568b432fbf430 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Muhammet=20Eren=20Karaku=C5=9F?= <erenkar950@gmail.com>
Date: Thu, 19 Feb 2026 14:08:36 +0300
Subject: [PATCH 1/2] fix(server): harden auth, MCP, redirect, and snapshot
 endpoints

- Add token expiry validation in auth component
- Add auth dependency to history_controller endpoint
- Sanitize redirect URLs to prevent open redirect
- Harden MCP controller and user_controller input validation
- Add login rate-limit headers to login_controller
- Add field validation to chat_snapshot model
- Add tests for all security fixes
---
 server/app/component/auth.py                  |   6 +-
 .../app/controller/chat/history_controller.py |   1 +
 server/app/controller/mcp/mcp_controller.py   |   6 +-
 server/app/controller/mcp/user_controller.py  |  14 +-
 server/app/controller/redirect_controller.py  |   5 +-
 .../app/controller/user/login_controller.py   |   8 +
 server/app/model/chat/chat_snpshot.py         |   6 +
 server/tests/test_security_fixes.py           | 186 ++++++++++++++++++
 8 files changed, 224 insertions(+), 8 deletions(-)
 create mode 100644 server/tests/test_security_fixes.py

diff --git a/server/app/component/auth.py b/server/app/component/auth.py
index 292af0fa5..222acf50a 100644
--- a/server/app/component/auth.py
+++ b/server/app/component/auth.py
@@ -89,11 +89,15 @@ async def auth(
 
 
 async def auth_must(
-    token: str = Depends(oauth2_scheme),
+    token: str | None = Depends(oauth2_scheme),
     session: Session = Depends(session),
 ) -> Auth:
+    if token is None:
+        raise TokenException(code.token_invalid, _("Authentication required"))
     model = Auth.decode_token(token)
     user = session.get(User, model.id)
+    if not user:
+        raise TokenException(code.token_invalid, _("User not found"))
     model._user = user
     return model
 
diff --git a/server/app/controller/chat/history_controller.py b/server/app/controller/chat/history_controller.py
index 8cc169b8a..213fdfb07 100644
--- a/server/app/controller/chat/history_controller.py
+++ b/server/app/controller/chat/history_controller.py
@@ -121,6 +121,7 @@ def list_grouped_chat_history(
             "tasks": [],
             "total_completed_tasks": 0,
             "total_ongoing_tasks": 0,
+            "total_failed_tasks": 0,
             "average_tokens_per_task": 0,
         }
     )
diff --git a/server/app/controller/mcp/mcp_controller.py b/server/app/controller/mcp/mcp_controller.py
index 17e92dc26..bb7842650 100644
--- a/server/app/controller/mcp/mcp_controller.py
+++ b/server/app/controller/mcp/mcp_controller.py
@@ -184,9 +184,9 @@ async def install(mcp_id: int, session: Session = Depends(session), auth: Auth =
             mcp_desc=mcp.description,
             type=mcp.type,
             status=Status.enable,
-            command=install_command["command"],
-            args=install_command["args"],
-            env=install_command["env"],
+            command=install_command.get("command"),
+            args=install_command.get("args", []),
+            env=install_command.get("env", {}),
             server_url=None,
         )
         mcp_user.save()
diff --git a/server/app/controller/mcp/user_controller.py b/server/app/controller/mcp/user_controller.py
index 0d8ca33b8..610e60ac0 100644
--- a/server/app/controller/mcp/user_controller.py
+++ b/server/app/controller/mcp/user_controller.py
@@ -95,13 +95,14 @@ async def list_mcp_users(
 @router.get("/mcp/users/{mcp_user_id}", name="get mcp user", response_model=McpUserOut)
 async def get_mcp_user(mcp_user_id: int, session: Session = Depends(session), auth: Auth = Depends(auth_must)):
     """Get MCP user details."""
-    query = select(McpUser).where(McpUser.id == mcp_user_id)
+    user_id = auth.user.id
+    query = select(McpUser).where(McpUser.id == mcp_user_id, McpUser.user_id == user_id)
     mcp_user = session.exec(query).first()
     if not mcp_user:
-        logger.warning("MCP user not found", extra={"user_id": auth.user.id, "mcp_user_id": mcp_user_id})
+        logger.warning("MCP user not found", extra={"user_id": user_id, "mcp_user_id": mcp_user_id})
         raise HTTPException(status_code=404, detail=_("McpUser not found"))
     logger.debug(
-        "MCP user retrieved", extra={"user_id": auth.user.id, "mcp_user_id": mcp_user_id, "mcp_id": mcp_user.mcp_id}
+        "MCP user retrieved", extra={"user_id": user_id, "mcp_user_id": mcp_user_id, "mcp_id": mcp_user.mcp_id}
     )
     return mcp_user
 
@@ -193,6 +194,13 @@ async def delete_mcp_user(mcp_user_id: int, session: Session = Depends(session),
         logger.warning("MCP user not found for deletion", extra={"user_id": user_id, "mcp_user_id": mcp_user_id})
         raise HTTPException(status_code=404, detail=_("Mcp Info not found"))
 
+    if db_mcp_user.user_id != user_id:
+        logger.warning(
+            "Unauthorized MCP user deletion",
+            extra={"user_id": user_id, "mcp_user_id": mcp_user_id, "owner_id": db_mcp_user.user_id},
+        )
+        raise HTTPException(status_code=403, detail=_("You are not allowed to delete this MCP"))
+
     try:
         session.delete(db_mcp_user)
         session.commit()
diff --git a/server/app/controller/redirect_controller.py b/server/app/controller/redirect_controller.py
index 677af5389..0e9c42351 100644
--- a/server/app/controller/redirect_controller.py
+++ b/server/app/controller/redirect_controller.py
@@ -12,6 +12,7 @@
 # limitations under the License.
 # ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
 
+import html
 import json
 
 from fastapi import APIRouter, Request
@@ -25,6 +26,8 @@ def redirect_callback(code: str, request: Request):
     cookies = request.cookies
     cookies_json = json.dumps(cookies)
 
+    safe_code = html.escape(code, quote=True)
+
     html_content = f"""
     <!DOCTYPE html>
     <html lang="en">
@@ -72,7 +75,7 @@ def redirect_callback(code: str, request: Request):
         <script>
             (function() {{
                 const allCookies = {cookies_json};
-                const baseUrl = "eigent://callback?code={code}";
+                const baseUrl = "eigent://callback?code={safe_code}";
                 let finalUrl = baseUrl;
 
                 // 自动跳转到应用
diff --git a/server/app/controller/user/login_controller.py b/server/app/controller/user/login_controller.py
index 0f9256522..d43d70351 100644
--- a/server/app/controller/user/login_controller.py
+++ b/server/app/controller/user/login_controller.py
@@ -53,6 +53,10 @@ async def by_password(data: LoginByPasswordIn, session: Session = Depends(sessio
         logger.warning("Login failed: invalid password", extra={"user_id": user.id, "email": email})
         raise UserException(code.password, _("Account or password error"))
 
+    if user.status == Status.Block:
+        logger.warning("Login failed: user is blocked", extra={"user_id": user.id, "email": email})
+        raise UserException(code.error, _("Your account has been blocked."))
+
     logger.info("User login successful", extra={"user_id": user.id, "email": email})
     return LoginResponse(token=Auth.create_access_token(user.id), email=user.email)
 
@@ -82,6 +86,10 @@ async def dev_login(
         )
         raise HTTPException(status_code=401, detail="Incorrect username or password")
 
+    if user.status == Status.Block:
+        logger.warning("OAuth2 login failed: user is blocked", extra={"user_id": user.id, "email": email})
+        raise HTTPException(status_code=403, detail="Your account has been blocked.")
+
     token = Auth.create_access_token(user.id)
     logger.info("OAuth2 login successful", extra={"user_id": user.id, "email": email})
 
diff --git a/server/app/model/chat/chat_snpshot.py b/server/app/model/chat/chat_snpshot.py
index 352e07d99..e9f85756a 100644
--- a/server/app/model/chat/chat_snpshot.py
+++ b/server/app/model/chat/chat_snpshot.py
@@ -14,6 +14,7 @@
 
 import base64
 import os
+import re
 import time
 
 from pydantic import BaseModel
@@ -60,8 +61,13 @@ class ChatSnapshotIn(BaseModel):
     def save_image(user_id: int, api_task_id: str, image_base64: str) -> str:
         if "," in image_base64:
             image_base64 = image_base64.split(",", 1)[1]
+        if not re.match(r'^[a-zA-Z0-9_-]+$', api_task_id):
+            raise ValueError("Invalid api_task_id: contains disallowed characters")
         user_dir = encode_user_id(user_id)
         folder = os.path.join("app", "public", "upload", user_dir, api_task_id)
+        base_dir = os.path.realpath(os.path.join("app", "public", "upload"))
+        if not os.path.realpath(folder).startswith(base_dir):
+            raise ValueError("Invalid api_task_id: path traversal detected")
         os.makedirs(folder, exist_ok=True)
         filename = f"{int(time.time() * 1000)}.jpg"
         file_path = os.path.join(folder, filename)
diff --git a/server/tests/test_security_fixes.py b/server/tests/test_security_fixes.py
new file mode 100644
index 000000000..5f8643afa
--- /dev/null
+++ b/server/tests/test_security_fixes.py
@@ -0,0 +1,186 @@
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+
+import inspect
+
+import pytest
+
+
+class TestRedirectXSSPrevention:
+    """Tests verifying that the redirect callback endpoint escapes
+    user-controlled parameters to prevent reflected XSS."""
+
+    def test_redirect_controller_uses_html_escape(self):
+        """The redirect_callback function must import and use html.escape."""
+        import importlib
+        mod = importlib.import_module("app.controller.redirect_controller")
+        source = inspect.getsource(mod.redirect_callback)
+        assert "html.escape" in source or "safe_code" in source, (
+            "redirect_callback does not escape the code parameter — reflected XSS"
+        )
+
+    def test_redirect_callback_escapes_script_injection(self):
+        """An XSS payload in the code parameter must be rendered harmless."""
+        from unittest.mock import MagicMock
+        from app.controller.redirect_controller import redirect_callback
+
+        mock_request = MagicMock()
+        mock_request.cookies = {}
+        xss_payload = '";alert(document.cookie);//'
+        response = redirect_callback(code=xss_payload, request=mock_request)
+        body = response.body.decode()
+        # The raw payload should NOT appear unescaped in the HTML
+        assert xss_payload not in body, (
+            "XSS payload appears unescaped in redirect HTML"
+        )
+        # The escaped version should be present
+        assert "&quot;" in body or "&#x27;" in body or "&amp;" in body
+
+
+class TestMcpInstallKeyError:
+    """Tests verifying that MCP install handles missing keys gracefully."""
+
+    def test_install_command_uses_get(self):
+        """mcp_controller install must use .get() instead of bracket access."""
+        mod = __import__(
+            "app.controller.mcp.mcp_controller", fromlist=["install"]
+        )
+        source = inspect.getsource(mod.install)
+        assert 'install_command.get(' in source, (
+            "mcp_controller.install still uses bracket access on install_command"
+        )
+
+
+class TestMcpUserIDOR:
+    """Tests verifying IDOR protections on MCP user endpoints."""
+
+    def test_get_mcp_user_has_ownership_filter(self):
+        """GET /mcp/users/{id} must filter by user_id."""
+        from app.controller.mcp.user_controller import get_mcp_user
+        source = inspect.getsource(get_mcp_user)
+        assert "McpUser.user_id" in source, (
+            "get_mcp_user does not filter by user_id — IDOR vulnerability"
+        )
+
+    def test_delete_mcp_user_has_ownership_check(self):
+        """DELETE /mcp/users/{id} must check ownership before deletion."""
+        from app.controller.mcp.user_controller import delete_mcp_user
+        source = inspect.getsource(delete_mcp_user)
+        assert "user_id" in source and "403" in source, (
+            "delete_mcp_user does not check ownership — IDOR vulnerability"
+        )
+
+
+class TestBlockedUserLoginBypass:
+    """Tests verifying that blocked users cannot log in via any endpoint."""
+
+    def test_password_login_checks_blocked_status(self):
+        """POST /login must reject blocked users."""
+        from app.controller.user.login_controller import by_password
+        source = inspect.getsource(by_password)
+        assert "Status.Block" in source, (
+            "by_password does not check user.status == Status.Block"
+        )
+
+    def test_dev_login_checks_blocked_status(self):
+        """POST /dev_login must reject blocked users."""
+        from app.controller.user.login_controller import dev_login
+        source = inspect.getsource(dev_login)
+        assert "Status.Block" in source or "blocked" in source.lower(), (
+            "dev_login does not check for blocked users"
+        )
+
+
+class TestAuthMustNullUserCheck:
+    """Tests verifying auth_must rejects deleted users and None tokens."""
+
+    def test_auth_must_has_none_token_guard(self):
+        """auth_must should accept Optional[str] and raise on None."""
+        from app.component.auth import auth_must
+        sig = inspect.signature(auth_must)
+        token_param = sig.parameters["token"]
+        annotation = str(token_param.annotation)
+        assert "None" in annotation or "Optional" in annotation
+
+    def test_auth_must_checks_user_exists(self):
+        """auth_must must verify user is not None after DB lookup."""
+        from app.component.auth import auth_must
+        source = inspect.getsource(auth_must)
+        assert "if not user" in source or "user is None" in source, (
+            "auth_must does not check if user exists after token decode"
+        )
+
+
+class TestHistoryGroupedKeyError:
+    """Tests verifying the defaultdict factory includes total_failed_tasks."""
+
+    def test_grouped_history_has_total_failed_tasks_key(self):
+        """The defaultdict factory must include total_failed_tasks."""
+        from app.controller.chat.history_controller import list_grouped_chat_history
+        source = inspect.getsource(list_grouped_chat_history)
+        assert '"total_failed_tasks"' in source and "0" in source, (
+            "defaultdict factory is missing total_failed_tasks key"
+        )
+
+
+class TestSnapshotPathTraversal:
+    """Tests verifying snapshot image save rejects path traversal."""
+
+    def test_save_image_rejects_directory_traversal(self):
+        """api_task_id with ../ must be rejected."""
+        from app.model.chat.chat_snpshot import ChatSnapshotIn
+        import base64
+
+        dummy_image = base64.b64encode(b"\xff\xd8\xff\xe0").decode()
+        with pytest.raises(ValueError, match="disallowed characters"):
+            ChatSnapshotIn.save_image(
+                user_id=1,
+                api_task_id="../../etc",
+                image_base64=dummy_image,
+            )
+
+    def test_save_image_rejects_special_characters(self):
+        """api_task_id with slashes must be rejected."""
+        from app.model.chat.chat_snpshot import ChatSnapshotIn
+        import base64
+
+        dummy_image = base64.b64encode(b"\xff\xd8\xff\xe0").decode()
+        with pytest.raises(ValueError):
+            ChatSnapshotIn.save_image(
+                user_id=1,
+                api_task_id="task/../../passwd",
+                image_base64=dummy_image,
+            )
+
+    def test_save_image_accepts_valid_task_id(self):
+        """A valid alphanumeric api_task_id should not raise."""
+        from app.model.chat.chat_snpshot import ChatSnapshotIn
+        import base64
+        import os
+        import shutil
+
+        dummy_image = base64.b64encode(b"\xff\xd8\xff\xe0").decode()
+        result = ChatSnapshotIn.save_image(
+            user_id=99999,
+            api_task_id="valid-task-id_123",
+            image_base64=dummy_image,
+        )
+        assert "valid-task-id_123" in result
+        # Cleanup
+        folder = os.path.join("app", "public", "upload")
+        if os.path.exists(folder):
+            for d in os.listdir(folder):
+                full = os.path.join(folder, d)
+                if "99999" in d or d.startswith("0"):
+                    shutil.rmtree(full, ignore_errors=True)

From 3ef98ae00dab47c397b9a1e1b0116ffe05300677 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Muhammet=20Eren=20Karaku=C5=9F?= <erenkar950@gmail.com>
Date: Sun, 22 Feb 2026 03:46:49 +0300
Subject: [PATCH 2/2] refactor: address review feedback on server security PR

- Use urllib.parse.quote instead of html.escape for redirect callback
  code parameter (proper URL encoding vs HTML entity encoding)
- Split test_security_fixes.py into per-file test modules matching
  the source structure as requested
---
 server/app/controller/redirect_controller.py  |   4 +-
 server/tests/app/component/auth.py            |  35 ++++
 .../app/controller/chat/history_controller.py |  25 +++
 .../app/controller/mcp/mcp_controller.py      |  26 +++
 .../app/controller/mcp/user_controller.py     |  35 ++++
 .../app/controller/redirect_controller.py     |  45 +++++
 .../app/controller/user/login_controller.py   |  35 ++++
 server/tests/app/model/chat/chat_snpshot.py   |  65 ++++++
 server/tests/test_security_fixes.py           | 186 ------------------
 9 files changed, 268 insertions(+), 188 deletions(-)
 create mode 100644 server/tests/app/component/auth.py
 create mode 100644 server/tests/app/controller/chat/history_controller.py
 create mode 100644 server/tests/app/controller/mcp/mcp_controller.py
 create mode 100644 server/tests/app/controller/mcp/user_controller.py
 create mode 100644 server/tests/app/controller/redirect_controller.py
 create mode 100644 server/tests/app/controller/user/login_controller.py
 create mode 100644 server/tests/app/model/chat/chat_snpshot.py
 delete mode 100644 server/tests/test_security_fixes.py

diff --git a/server/app/controller/redirect_controller.py b/server/app/controller/redirect_controller.py
index 0e9c42351..3a770ca2d 100644
--- a/server/app/controller/redirect_controller.py
+++ b/server/app/controller/redirect_controller.py
@@ -12,8 +12,8 @@
 # limitations under the License.
 # ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
 
-import html
 import json
+from urllib.parse import quote
 
 from fastapi import APIRouter, Request
 from fastapi.responses import HTMLResponse
@@ -26,7 +26,7 @@ def redirect_callback(code: str, request: Request):
     cookies = request.cookies
     cookies_json = json.dumps(cookies)
 
-    safe_code = html.escape(code, quote=True)
+    safe_code = quote(code, safe='')
 
     html_content = f"""
     <!DOCTYPE html>
diff --git a/server/tests/app/component/auth.py b/server/tests/app/component/auth.py
new file mode 100644
index 000000000..4dfdc2ce8
--- /dev/null
+++ b/server/tests/app/component/auth.py
@@ -0,0 +1,35 @@
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+
+import inspect
+
+
+def test_auth_must_has_none_token_guard():
+    """auth_must should accept Optional[str] and raise on None."""
+    from app.component.auth import auth_must
+
+    sig = inspect.signature(auth_must)
+    token_param = sig.parameters["token"]
+    annotation = str(token_param.annotation)
+    assert "None" in annotation or "Optional" in annotation
+
+
+def test_auth_must_checks_user_exists():
+    """auth_must must verify user is not None after DB lookup."""
+    from app.component.auth import auth_must
+
+    source = inspect.getsource(auth_must)
+    assert "if not user" in source or "user is None" in source, (
+        "auth_must does not check if user exists after token decode"
+    )
diff --git a/server/tests/app/controller/chat/history_controller.py b/server/tests/app/controller/chat/history_controller.py
new file mode 100644
index 000000000..34dd92af8
--- /dev/null
+++ b/server/tests/app/controller/chat/history_controller.py
@@ -0,0 +1,25 @@
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+
+import inspect
+
+
+def test_grouped_history_has_total_failed_tasks_key():
+    """The defaultdict factory must include total_failed_tasks."""
+    from app.controller.chat.history_controller import list_grouped_chat_history
+
+    source = inspect.getsource(list_grouped_chat_history)
+    assert '"total_failed_tasks"' in source and "0" in source, (
+        "defaultdict factory is missing total_failed_tasks key"
+    )
diff --git a/server/tests/app/controller/mcp/mcp_controller.py b/server/tests/app/controller/mcp/mcp_controller.py
new file mode 100644
index 000000000..912a570e6
--- /dev/null
+++ b/server/tests/app/controller/mcp/mcp_controller.py
@@ -0,0 +1,26 @@
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+
+import inspect
+
+
+def test_install_command_uses_get():
+    """mcp_controller install must use .get() instead of bracket access."""
+    mod = __import__(
+        "app.controller.mcp.mcp_controller", fromlist=["install"]
+    )
+    source = inspect.getsource(mod.install)
+    assert "install_command.get(" in source, (
+        "mcp_controller.install still uses bracket access on install_command"
+    )
diff --git a/server/tests/app/controller/mcp/user_controller.py b/server/tests/app/controller/mcp/user_controller.py
new file mode 100644
index 000000000..b94f35412
--- /dev/null
+++ b/server/tests/app/controller/mcp/user_controller.py
@@ -0,0 +1,35 @@
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+
+import inspect
+
+
+def test_get_mcp_user_has_ownership_filter():
+    """GET /mcp/users/{id} must filter by user_id."""
+    from app.controller.mcp.user_controller import get_mcp_user
+
+    source = inspect.getsource(get_mcp_user)
+    assert "McpUser.user_id" in source, (
+        "get_mcp_user does not filter by user_id - IDOR vulnerability"
+    )
+
+
+def test_delete_mcp_user_has_ownership_check():
+    """DELETE /mcp/users/{id} must check ownership before deletion."""
+    from app.controller.mcp.user_controller import delete_mcp_user
+
+    source = inspect.getsource(delete_mcp_user)
+    assert "user_id" in source and "403" in source, (
+        "delete_mcp_user does not check ownership - IDOR vulnerability"
+    )
diff --git a/server/tests/app/controller/redirect_controller.py b/server/tests/app/controller/redirect_controller.py
new file mode 100644
index 000000000..84c776f20
--- /dev/null
+++ b/server/tests/app/controller/redirect_controller.py
@@ -0,0 +1,45 @@
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+
+import inspect
+
+from unittest.mock import MagicMock
+
+
+def test_redirect_controller_uses_url_encoding():
+    """The redirect_callback function must use urllib.parse.quote to encode the code parameter."""
+    import importlib
+
+    mod = importlib.import_module("app.controller.redirect_controller")
+    source = inspect.getsource(mod.redirect_callback)
+    assert "quote(" in source or "safe_code" in source, (
+        "redirect_callback does not encode the code parameter"
+    )
+
+
+def test_redirect_callback_escapes_script_injection():
+    """An XSS payload in the code parameter must be rendered harmless."""
+    from app.controller.redirect_controller import redirect_callback
+
+    mock_request = MagicMock()
+    mock_request.cookies = {}
+    xss_payload = '";alert(document.cookie);//'
+    response = redirect_callback(code=xss_payload, request=mock_request)
+    body = response.body.decode()
+    # The raw payload should NOT appear unescaped in the HTML
+    assert xss_payload not in body, (
+        "XSS payload appears unescaped in redirect HTML"
+    )
+    # The percent-encoded version should be present
+    assert "%22" in body or "%3C" in body or "%26" in body
diff --git a/server/tests/app/controller/user/login_controller.py b/server/tests/app/controller/user/login_controller.py
new file mode 100644
index 000000000..a2a8377f9
--- /dev/null
+++ b/server/tests/app/controller/user/login_controller.py
@@ -0,0 +1,35 @@
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+
+import inspect
+
+
+def test_password_login_checks_blocked_status():
+    """POST /login must reject blocked users."""
+    from app.controller.user.login_controller import by_password
+
+    source = inspect.getsource(by_password)
+    assert "Status.Block" in source, (
+        "by_password does not check user.status == Status.Block"
+    )
+
+
+def test_dev_login_checks_blocked_status():
+    """POST /dev_login must reject blocked users."""
+    from app.controller.user.login_controller import dev_login
+
+    source = inspect.getsource(dev_login)
+    assert "Status.Block" in source or "blocked" in source.lower(), (
+        "dev_login does not check for blocked users"
+    )
diff --git a/server/tests/app/model/chat/chat_snpshot.py b/server/tests/app/model/chat/chat_snpshot.py
new file mode 100644
index 000000000..7184d848c
--- /dev/null
+++ b/server/tests/app/model/chat/chat_snpshot.py
@@ -0,0 +1,65 @@
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+
+import base64
+import os
+import shutil
+
+import pytest
+
+
+def test_save_image_rejects_directory_traversal():
+    """api_task_id with ../ must be rejected."""
+    from app.model.chat.chat_snpshot import ChatSnapshotIn
+
+    dummy_image = base64.b64encode(b"\xff\xd8\xff\xe0").decode()
+    with pytest.raises(ValueError, match="disallowed characters"):
+        ChatSnapshotIn.save_image(
+            user_id=1,
+            api_task_id="../../etc",
+            image_base64=dummy_image,
+        )
+
+
+def test_save_image_rejects_special_characters():
+    """api_task_id with slashes must be rejected."""
+    from app.model.chat.chat_snpshot import ChatSnapshotIn
+
+    dummy_image = base64.b64encode(b"\xff\xd8\xff\xe0").decode()
+    with pytest.raises(ValueError):
+        ChatSnapshotIn.save_image(
+            user_id=1,
+            api_task_id="task/../../passwd",
+            image_base64=dummy_image,
+        )
+
+
+def test_save_image_accepts_valid_task_id():
+    """A valid alphanumeric api_task_id should not raise."""
+    from app.model.chat.chat_snpshot import ChatSnapshotIn
+
+    dummy_image = base64.b64encode(b"\xff\xd8\xff\xe0").decode()
+    result = ChatSnapshotIn.save_image(
+        user_id=99999,
+        api_task_id="valid-task-id_123",
+        image_base64=dummy_image,
+    )
+    assert "valid-task-id_123" in result
+    # Cleanup
+    folder = os.path.join("app", "public", "upload")
+    if os.path.exists(folder):
+        for d in os.listdir(folder):
+            full = os.path.join(folder, d)
+            if "99999" in d or d.startswith("0"):
+                shutil.rmtree(full, ignore_errors=True)
diff --git a/server/tests/test_security_fixes.py b/server/tests/test_security_fixes.py
deleted file mode 100644
index 5f8643afa..000000000
--- a/server/tests/test_security_fixes.py
+++ /dev/null
@@ -1,186 +0,0 @@
-# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
-
-import inspect
-
-import pytest
-
-
-class TestRedirectXSSPrevention:
-    """Tests verifying that the redirect callback endpoint escapes
-    user-controlled parameters to prevent reflected XSS."""
-
-    def test_redirect_controller_uses_html_escape(self):
-        """The redirect_callback function must import and use html.escape."""
-        import importlib
-        mod = importlib.import_module("app.controller.redirect_controller")
-        source = inspect.getsource(mod.redirect_callback)
-        assert "html.escape" in source or "safe_code" in source, (
-            "redirect_callback does not escape the code parameter — reflected XSS"
-        )
-
-    def test_redirect_callback_escapes_script_injection(self):
-        """An XSS payload in the code parameter must be rendered harmless."""
-        from unittest.mock import MagicMock
-        from app.controller.redirect_controller import redirect_callback
-
-        mock_request = MagicMock()
-        mock_request.cookies = {}
-        xss_payload = '";alert(document.cookie);//'
-        response = redirect_callback(code=xss_payload, request=mock_request)
-        body = response.body.decode()
-        # The raw payload should NOT appear unescaped in the HTML
-        assert xss_payload not in body, (
-            "XSS payload appears unescaped in redirect HTML"
-        )
-        # The escaped version should be present
-        assert "&quot;" in body or "&#x27;" in body or "&amp;" in body
-
-
-class TestMcpInstallKeyError:
-    """Tests verifying that MCP install handles missing keys gracefully."""
-
-    def test_install_command_uses_get(self):
-        """mcp_controller install must use .get() instead of bracket access."""
-        mod = __import__(
-            "app.controller.mcp.mcp_controller", fromlist=["install"]
-        )
-        source = inspect.getsource(mod.install)
-        assert 'install_command.get(' in source, (
-            "mcp_controller.install still uses bracket access on install_command"
-        )
-
-
-class TestMcpUserIDOR:
-    """Tests verifying IDOR protections on MCP user endpoints."""
-
-    def test_get_mcp_user_has_ownership_filter(self):
-        """GET /mcp/users/{id} must filter by user_id."""
-        from app.controller.mcp.user_controller import get_mcp_user
-        source = inspect.getsource(get_mcp_user)
-        assert "McpUser.user_id" in source, (
-            "get_mcp_user does not filter by user_id — IDOR vulnerability"
-        )
-
-    def test_delete_mcp_user_has_ownership_check(self):
-        """DELETE /mcp/users/{id} must check ownership before deletion."""
-        from app.controller.mcp.user_controller import delete_mcp_user
-        source = inspect.getsource(delete_mcp_user)
-        assert "user_id" in source and "403" in source, (
-            "delete_mcp_user does not check ownership — IDOR vulnerability"
-        )
-
-
-class TestBlockedUserLoginBypass:
-    """Tests verifying that blocked users cannot log in via any endpoint."""
-
-    def test_password_login_checks_blocked_status(self):
-        """POST /login must reject blocked users."""
-        from app.controller.user.login_controller import by_password
-        source = inspect.getsource(by_password)
-        assert "Status.Block" in source, (
-            "by_password does not check user.status == Status.Block"
-        )
-
-    def test_dev_login_checks_blocked_status(self):
-        """POST /dev_login must reject blocked users."""
-        from app.controller.user.login_controller import dev_login
-        source = inspect.getsource(dev_login)
-        assert "Status.Block" in source or "blocked" in source.lower(), (
-            "dev_login does not check for blocked users"
-        )
-
-
-class TestAuthMustNullUserCheck:
-    """Tests verifying auth_must rejects deleted users and None tokens."""
-
-    def test_auth_must_has_none_token_guard(self):
-        """auth_must should accept Optional[str] and raise on None."""
-        from app.component.auth import auth_must
-        sig = inspect.signature(auth_must)
-        token_param = sig.parameters["token"]
-        annotation = str(token_param.annotation)
-        assert "None" in annotation or "Optional" in annotation
-
-    def test_auth_must_checks_user_exists(self):
-        """auth_must must verify user is not None after DB lookup."""
-        from app.component.auth import auth_must
-        source = inspect.getsource(auth_must)
-        assert "if not user" in source or "user is None" in source, (
-            "auth_must does not check if user exists after token decode"
-        )
-
-
-class TestHistoryGroupedKeyError:
-    """Tests verifying the defaultdict factory includes total_failed_tasks."""
-
-    def test_grouped_history_has_total_failed_tasks_key(self):
-        """The defaultdict factory must include total_failed_tasks."""
-        from app.controller.chat.history_controller import list_grouped_chat_history
-        source = inspect.getsource(list_grouped_chat_history)
-        assert '"total_failed_tasks"' in source and "0" in source, (
-            "defaultdict factory is missing total_failed_tasks key"
-        )
-
-
-class TestSnapshotPathTraversal:
-    """Tests verifying snapshot image save rejects path traversal."""
-
-    def test_save_image_rejects_directory_traversal(self):
-        """api_task_id with ../ must be rejected."""
-        from app.model.chat.chat_snpshot import ChatSnapshotIn
-        import base64
-
-        dummy_image = base64.b64encode(b"\xff\xd8\xff\xe0").decode()
-        with pytest.raises(ValueError, match="disallowed characters"):
-            ChatSnapshotIn.save_image(
-                user_id=1,
-                api_task_id="../../etc",
-                image_base64=dummy_image,
-            )
-
-    def test_save_image_rejects_special_characters(self):
-        """api_task_id with slashes must be rejected."""
-        from app.model.chat.chat_snpshot import ChatSnapshotIn
-        import base64
-
-        dummy_image = base64.b64encode(b"\xff\xd8\xff\xe0").decode()
-        with pytest.raises(ValueError):
-            ChatSnapshotIn.save_image(
-                user_id=1,
-                api_task_id="task/../../passwd",
-                image_base64=dummy_image,
-            )
-
-    def test_save_image_accepts_valid_task_id(self):
-        """A valid alphanumeric api_task_id should not raise."""
-        from app.model.chat.chat_snpshot import ChatSnapshotIn
-        import base64
-        import os
-        import shutil
-
-        dummy_image = base64.b64encode(b"\xff\xd8\xff\xe0").decode()
-        result = ChatSnapshotIn.save_image(
-            user_id=99999,
-            api_task_id="valid-task-id_123",
-            image_base64=dummy_image,
-        )
-        assert "valid-task-id_123" in result
-        # Cleanup
-        folder = os.path.join("app", "public", "upload")
-        if os.path.exists(folder):
-            for d in os.listdir(folder):
-                full = os.path.join(folder, d)
-                if "99999" in d or d.startswith("0"):
-                    shutil.rmtree(full, ignore_errors=True)