feat(security): add security config UI and improve auth/MCP security

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: EstrellaXD

CHANGELOG.md

@@ -1,3 +1,41 @@
+# [Unreleased]
+
+## Backend
+
+### Added
+
+- 新增 `Security` 配置模型，支持登录 IP 白名单、MCP IP 白名单和 Bearer Token 认证
+- 新增登录端点 IP 白名单检查中间件 (`check_login_ip`)
+- MCP 安全中间件升级为可配置模式：支持 CIDR 白名单 + Bearer Token 双重认证
+- 认证端点支持 `Authorization: Bearer` 令牌绕过 Cookie 登录
+- 配置 API `_sanitize_dict` 修复：仅对字符串值进行脱敏，避免误处理非字符串字段
+
+- 新增番剧放送日手动设置 API (`PATCH /api/v1/bangumi/{id}/weekday`)，支持锁定放送日防止日历刷新覆盖
+- 数据库迁移 v9：`bangumi` 表新增 `weekday_locked` 列
+
+### Changed
+
+- 重构认证模块：提取 `_issue_token` 公共方法，消除 3 处重复的 JWT 签发逻辑
+- `get_current_user` 简化为三级认证（DEV 绕过 → Bearer Token → Cookie JWT）
+- `LocalNetworkMiddleware` 重命名为 `McpAccessMiddleware`，从硬编码 RFC 1918 改为读取配置
+
+### Tests
+
+- 新增 101 个单元测试覆盖安全、认证、配置、下载器和 MockDownloader 模块
+
+## Frontend
+
+### Added
+
+- 新增日历拖拽排列功能：可将「未知」番剧拖入星期列，自动设置放送日并锁定
+  - 拖入后显示紫色图钉图标，鼠标悬停显示取消按钮
+  - 锁定的番剧在日历刷新时不会被覆盖
+  - 使用 vuedraggable 实现流畅拖拽动画
+- 新增安全设置组件 (`config-security.vue`)，支持在 WebUI 中配置 IP 白名单和 Token
+- 前端 `Security` 类型定义和初始化配置
+
+---
+
 # [3.2.3] - 2026-02-23
 
 ## Backend

backend/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "auto-bangumi"
-version = "3.2.3"
+version = "3.2.4"
 description = "AutoBangumi - Automated anime download manager"
 requires-python = ">=3.13"
 dependencies = [

backend/src/module/api/auth.py

@@ -9,6 +9,7 @@
 from module.security.api import (
     active_user,
     auth_user,
+    check_login_ip,
     get_current_user,
     update_user_info,
 )
@@ -18,42 +19,49 @@
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 
+_TOKEN_EXPIRY_DAYS = 1
+_TOKEN_MAX_AGE = 86400
 
-@router.post("/login", response_model=dict)
+
+def _issue_token(username: str, response: Response) -> dict:
+    """Create a JWT, set it as an HttpOnly cookie, and return the bearer payload."""
+    token = create_access_token(
+        data={"sub": username}, expires_delta=timedelta(days=_TOKEN_EXPIRY_DAYS)
+    )
+    response.set_cookie(key="token", value=token, httponly=True, max_age=_TOKEN_MAX_AGE)
+    return {"access_token": token, "token_type": "bearer"}
+
+
+@router.post("/login", response_model=dict, dependencies=[Depends(check_login_ip)])
 async def login(response: Response, form_data=Depends(OAuth2PasswordRequestForm)):
+    """Authenticate with username/password and issue a session token."""
     user = User(username=form_data.username, password=form_data.password)
     resp = auth_user(user)
     if resp.status:
-        token = create_access_token(
-            data={"sub": user.username}, expires_delta=timedelta(days=1)
-        )
-        response.set_cookie(key="token", value=token, httponly=True, max_age=86400)
-        return {"access_token": token, "token_type": "bearer"}
+        return _issue_token(user.username, response)
     return u_response(resp)
 
 
 @router.get(
     "/refresh_token", response_model=dict, dependencies=[Depends(get_current_user)]
 )
 async def refresh(response: Response, token: str = Cookie(None)):
+    """Refresh the current session token and update the active-user timestamp."""
     payload = decode_token(token)
     username = payload.get("sub") if payload else None
     if not username:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized"
         )
     active_user[username] = datetime.now()
-    new_token = create_access_token(
-        data={"sub": username}, expires_delta=timedelta(days=1)
-    )
-    response.set_cookie(key="token", value=new_token, httponly=True, max_age=86400)
-    return {"access_token": new_token, "token_type": "bearer"}
+    return _issue_token(username, response)
 
 
 @router.get(
     "/logout", response_model=APIResponse, dependencies=[Depends(get_current_user)]
 )
 async def logout(response: Response, token: str = Cookie(None)):
+    """Invalidate the session and clear the token cookie."""
     payload = decode_token(token)
     username = payload.get("sub") if payload else None
     if username:
@@ -69,24 +77,12 @@ async def logout(response: Response, token: str = Cookie(None)):
 async def update_user(
     user_data: UserUpdate, response: Response, token: str = Cookie(None)
 ):
+    """Update credentials for the current user and re-issue a fresh token."""
     payload = decode_token(token)
     old_user = payload.get("sub") if payload else None
     if not old_user:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized"
         )
     if update_user_info(user_data, old_user):
-        token = create_access_token(
-            data={"sub": old_user}, expires_delta=timedelta(days=1)
-        )
-        response.set_cookie(
-            key="token",
-            value=token,
-            httponly=True,
-            max_age=86400,
-        )
-        return {
-            "access_token": token,
-            "token_type": "bearer",
-            "message": "update success",
-        }
+        return {**_issue_token(old_user, response), "message": "update success"}

backend/src/module/api/config.py

@@ -14,11 +14,12 @@
 
 
 def _sanitize_dict(d: dict) -> dict:
+    """Recursively mask string values whose keys contain sensitive keywords."""
     result = {}
     for k, v in d.items():
         if isinstance(v, dict):
             result[k] = _sanitize_dict(v)
-        elif any(s in k.lower() for s in _SENSITIVE_KEYS):
+        elif isinstance(v, str) and any(s in k.lower() for s in _SENSITIVE_KEYS):
             result[k] = "********"
         else:
             result[k] = v
@@ -27,13 +28,15 @@ def _sanitize_dict(d: dict) -> dict:
 
 @router.get("/get", dependencies=[Depends(get_current_user)])
 async def get_config():
+    """Return the current configuration with sensitive fields masked."""
     return _sanitize_dict(settings.dict())
 
 
 @router.patch(
     "/update", response_model=APIResponse, dependencies=[Depends(get_current_user)]
 )
 async def update_config(config: Config):
+    """Persist and reload configuration from the supplied payload."""
     try:
         settings.save(config_dict=config.dict())
         settings.load()

backend/src/module/conf/config.py

@@ -7,7 +7,7 @@
 
 from module.models.config import Config
 
-from .const import ENV_TO_ATTR
+from .const import DEFAULT_SETTINGS, ENV_TO_ATTR
 
 logger = logging.getLogger(__name__)
 CONFIG_ROOT = Path("config")
@@ -27,6 +27,15 @@
 
 
 class Settings(Config):
+    """Runtime configuration singleton.
+
+    On construction, loads from ``CONFIG_PATH`` if the file exists (and
+    immediately re-saves to apply any migrations), otherwise bootstraps
+    defaults from environment variables via ``init()``.
+
+    Use ``settings`` module-level instance rather than instantiating directly.
+    """
+
     def __init__(self):
         super().__init__()
         if CONFIG_PATH.exists():
@@ -36,6 +45,7 @@ def __init__(self):
             self.init()
 
     def load(self):
+        """Load and validate configuration from ``CONFIG_PATH``, applying migrations."""
         with open(CONFIG_PATH, "r", encoding="utf-8") as f:
             config = json.load(f)
         config = self._migrate_old_config(config)
@@ -65,20 +75,27 @@ def _migrate_old_config(config: dict) -> dict:
         for key in ("type", "custom_url", "token", "enable_tmdb"):
             rss_parser.pop(key, None)
 
+        # Add security section if missing (preserves local-network MCP default)
+        if "security" not in config:
+            config["security"] = DEFAULT_SETTINGS["security"]
+
         return config
 
     def save(self, config_dict: dict | None = None):
+        """Write configuration to ``CONFIG_PATH``. Uses current state when no dict supplied."""
         if not config_dict:
             config_dict = self.model_dump()
         with open(CONFIG_PATH, "w", encoding="utf-8") as f:
             json.dump(config_dict, f, indent=4, ensure_ascii=False)
 
     def init(self):
+        """Bootstrap a new config file from ``.env`` and environment variables."""
         load_dotenv(".env")
         self.__load_from_env()
         self.save()
 
     def __load_from_env(self):
+        """Apply ``ENV_TO_ATTR`` mappings from the process environment to the config dict."""
         config_dict = self.model_dump()
         for key, section in ENV_TO_ATTR.items():
             for env, attr in section.items():
@@ -97,12 +114,11 @@ def __load_from_env(self):
         logger.info("Config loaded from env")
 
     @staticmethod
-    def __val_from_env(env: str, attr: tuple):
+    def __val_from_env(env: str, attr: tuple | str):
+        """Return the environment variable value, applying the converter when attr is a tuple."""
         if isinstance(attr, tuple):
-            conv_func = attr[1]
-            return conv_func(os.environ[env])
-        else:
-            return os.environ[env]
+            return attr[1](os.environ[env])
+        return os.environ[env]
 
     @property
     def group_rules(self):

backend/src/module/conf/const.py

@@ -1,4 +1,8 @@
 # -*- encoding: utf-8 -*-
+# DEFAULT_SETTINGS: factory defaults written to config.json on first run.
+# ENV_TO_ATTR: maps AB_* environment variables to Config model attribute paths.
+#   Values are either a string attr name, a (attr_name, converter) tuple, or a
+#   list of such tuples when a single env var sets multiple attributes.
 DEFAULT_SETTINGS = {
     "program": {
         "rss_time": 900,
@@ -46,6 +50,20 @@
         "model": "gpt-3.5-turbo",
         "deployment_id": "",
     },
+    "security": {
+        "login_whitelist": [],
+        "login_tokens": [],
+        "mcp_whitelist": [
+            "127.0.0.0/8",
+            "10.0.0.0/8",
+            "172.16.0.0/12",
+            "192.168.0.0/16",
+            "::1/128",
+            "fe80::/10",
+            "fc00::/7",
+        ],
+        "mcp_tokens": [],
+    },
 }
 
 
@@ -99,8 +117,11 @@
 
 
 class BCOLORS:
+    """ANSI colour helpers for terminal output."""
+
     @staticmethod
     def _(color: str, *args: str) -> str:
+        """Wrap *args* in the given ANSI colour code and reset at the end."""
         strings = [str(s) for s in args]
         return f"{color}{', '.join(strings)}{BCOLORS.ENDC}"
 

backend/src/module/downloader/client/mock_downloader.py

@@ -29,10 +29,10 @@ def __init__(self):
             "rss_processing_enabled": True,
             "rss_refresh_interval": 30,
         }
-        logger.info("[MockDownloader] Initialized")
+        logger.debug("[MockDownloader] Initialized")
 
     async def auth(self, retry=3) -> bool:
-        logger.info("[MockDownloader] Auth successful (mocked)")
+        logger.debug("[MockDownloader] Auth successful (mocked)")
         return True
 
     async def logout(self):

backend/src/module/downloader/download_client.py

@@ -11,34 +11,42 @@
 
 
 class DownloadClient(TorrentPath):
+    """Unified async download client.
+
+    Wraps qBittorrent, Aria2, or MockDownloader behind a common interface.
+    Intended to be used as an async context manager; authentication is
+    performed on ``__aenter__`` and the session is closed on ``__aexit__``.
+    """
+
     def __init__(self):
         super().__init__()
         self.client = self.__getClient()
         self.authed = False
 
     @staticmethod
     def __getClient():
-        type = settings.downloader.type
+        """Instantiate the configured downloader client (qbittorrent | aria2 | mock)."""
+        downloader_type = settings.downloader.type
         host = settings.downloader.host
         username = settings.downloader.username
         password = settings.downloader.password
         ssl = settings.downloader.ssl
-        if type == "qbittorrent":
+        if downloader_type == "qbittorrent":
             from .client.qb_downloader import QbDownloader
 
             return QbDownloader(host, username, password, ssl)
-        elif type == "aria2":
+        elif downloader_type == "aria2":
             from .client.aria2_downloader import Aria2Downloader
 
             return Aria2Downloader(host, username, password)
-        elif type == "mock":
+        elif downloader_type == "mock":
             from .client.mock_downloader import MockDownloader
 
-            logger.info("[Downloader] Using MockDownloader for local development")
+            logger.debug("[Downloader] Using MockDownloader for local development")
             return MockDownloader()
         else:
-            logger.error(f"[Downloader] Unsupported downloader type: {type}")
-            raise Exception(f"Unsupported downloader type: {type}")
+            logger.error("[Downloader] Unsupported downloader type: %s", downloader_type)
+            raise Exception(f"Unsupported downloader type: {downloader_type}")
 
     async def __aenter__(self):
         if not self.authed:
@@ -65,6 +73,7 @@ async def check_host(self):
         return await self.client.check_host()
 
     async def init_downloader(self):
+        """Apply required qBittorrent RSS preferences and create the Bangumi category."""
         prefs = {
             "rss_auto_downloading_enabled": True,
             "rss_max_articles_per_feed": 500,
@@ -84,6 +93,7 @@ async def init_downloader(self):
             settings.downloader.path = self._join_path(prefs["save_path"], "Bangumi")
 
     async def set_rule(self, data: Bangumi):
+        """Create or update a qBittorrent RSS auto-download rule for one bangumi entry."""
         data.rule_name = self._rule_name(data)
         data.save_path = self._gen_save_path(data)
         rule = {
@@ -145,6 +155,12 @@ async def resume_torrent(self, hashes: str):
         await self.client.torrents_resume(hashes)
 
     async def add_torrent(self, torrent: Torrent | list, bangumi: Bangumi) -> bool:
+        """Download a torrent (or list of torrents) for the given bangumi entry.
+
+        Handles both magnet links and .torrent file URLs, fetching file bytes
+        when necessary. Tags each torrent with ``ab:<bangumi_id>`` for later
+        episode-offset lookup during rename.
+        """
         if not bangumi.save_path:
             bangumi.save_path = self._gen_save_path(bangumi)
         async with RequestContent() as req: