feat: configure admin session lifetime (#484)

Author: vastsa

apps/admin/dependencies.py

@@ -8,7 +8,12 @@
 import hmac
 import json
 import time
-from core.settings import settings
+from core.settings import (
+    ADMIN_SESSION_EXPIRE_DEFAULT,
+    ADMIN_SESSION_EXPIRE_MAX,
+    ADMIN_SESSION_EXPIRE_MIN,
+    settings,
+)
 from apps.admin.services import FileService, ConfigService, LocalFileService
 
 
@@ -19,17 +24,35 @@ def _get_jwt_secret() -> bytes:
     return secret.encode()
 
 
-def create_token(data: dict, expires_in: int = 3600 * 24 * 30) -> str:
+def get_admin_session_expire_seconds() -> int:
+    try:
+        expires_in = int(
+            getattr(settings, "adminSessionExpire", ADMIN_SESSION_EXPIRE_DEFAULT)
+        )
+    except (TypeError, ValueError):
+        return ADMIN_SESSION_EXPIRE_DEFAULT
+    if (
+        not ADMIN_SESSION_EXPIRE_MIN <= expires_in <= ADMIN_SESSION_EXPIRE_MAX
+        or expires_in % ADMIN_SESSION_EXPIRE_MIN != 0
+    ):
+        return ADMIN_SESSION_EXPIRE_DEFAULT
+    return expires_in
+
+
+def create_token(data: dict, expires_in: int | None = None) -> str:
     """
     创建JWT token
     :param data: 数据负载
     :param expires_in: 过期时间(秒)
     """
+    token_lifetime = (
+        get_admin_session_expire_seconds() if expires_in is None else expires_in
+    )
     header = base64.b64encode(
         json.dumps({"alg": "HS256", "typ": "JWT"}).encode()
     ).decode()
     payload = base64.b64encode(
-        json.dumps({**data, "exp": int(time.time()) + expires_in}).encode()
+        json.dumps({**data, "exp": int(time.time()) + token_lifetime}).encode()
     ).decode()
 
     signature = hmac.new(_get_jwt_secret(), f"{header}.{payload}".encode(), "sha256").digest()

apps/admin/services.py

@@ -6,7 +6,11 @@
 
 from core.response import APIResponse
 from core.storage import FileStorageInterface, storages
-from core.settings import settings
+from core.settings import (
+    ADMIN_SESSION_EXPIRE_MAX,
+    ADMIN_SESSION_EXPIRE_MIN,
+    settings,
+)
 from core.config import refresh_settings
 from core.security import INTERNAL_CONFIG_KEYS, generate_jwt_secret
 from apps.base.models import FileCodes, KeyValue
@@ -1496,6 +1500,7 @@ async def share_local_file(self, item):
 
 class ConfigService:
     INT_FIELDS = {
+        "adminSessionExpire",
         "enableChunk",
         "errorCount",
         "errorMinute",
@@ -1554,6 +1559,23 @@ async def update_config(self, data: dict):
             except (TypeError, ValueError):
                 raise HTTPException(status_code=400, detail=f"{key} 配置值格式错误")
 
+        try:
+            session_expire = int(next_config.get("adminSessionExpire"))
+        except (TypeError, ValueError):
+            raise HTTPException(
+                status_code=400,
+                detail="adminSessionExpire 配置值格式错误",
+            )
+        if (
+            not ADMIN_SESSION_EXPIRE_MIN <= session_expire <= ADMIN_SESSION_EXPIRE_MAX
+            or session_expire % ADMIN_SESSION_EXPIRE_MIN != 0
+        ):
+            raise HTTPException(
+                status_code=400,
+                detail="adminSessionExpire 必须是 1 到 365 个整天",
+            )
+        next_config["adminSessionExpire"] = session_expire
+
         if admin_password_changed:
             next_config["jwt_secret"] = generate_jwt_secret()
 

apps/admin/views.py

@@ -31,7 +31,11 @@
 )
 from core.response import APIResponse
 from apps.base.models import FileCodes, KeyValue
-from apps.admin.dependencies import create_token
+from apps.admin.dependencies import (
+    create_token,
+    get_admin_session_expire_seconds,
+    verify_token,
+)
 from core.settings import settings
 from core.utils import get_now, verify_password
 
@@ -53,8 +57,18 @@ async def login(data: LoginData):
     if not verify_password(data.password, settings.admin_token):
         raise HTTPException(status_code=401, detail="密码错误")
 
-    token = create_token({"is_admin": True})
-    return APIResponse(detail={"token": token, "token_type": "Bearer"})
+    expires_in = get_admin_session_expire_seconds()
+    token = create_token({"is_admin": True}, expires_in=expires_in)
+    return APIResponse(
+        detail={
+            "id": "admin",
+            "username": "admin",
+            "token": token,
+            "token_type": "Bearer",
+            "expires_at": verify_token(token)["exp"],
+            "expires_in": expires_in,
+        }
+    )
 
 
 @admin_api.get("/verify")

core/settings.py

@@ -4,6 +4,10 @@
 # @Software: PyCharm
 from pathlib import Path
 
+ADMIN_SESSION_EXPIRE_DEFAULT = 30 * 24 * 60 * 60
+ADMIN_SESSION_EXPIRE_MIN = 24 * 60 * 60
+ADMIN_SESSION_EXPIRE_MAX = 365 * 24 * 60 * 60
+
 BASE_DIR = Path(__file__).resolve().parent.parent
 data_root = BASE_DIR / "data"
 
@@ -40,6 +44,7 @@
     "webdav_proxy": 0,
     "admin_token": "",
     "jwt_secret": "",
+    "adminSessionExpire": ADMIN_SESSION_EXPIRE_DEFAULT,
     "openUpload": 1,
     "uploadSize": 1024 * 1024 * 10,
     "allowed_file_types": ["*"],

tests/test_admin_security.py

@@ -3,9 +3,13 @@
 import unittest
 
 import apps.admin.services as admin_services
+import apps.admin.views as admin_views
+import apps.admin.dependencies as admin_dependencies
 import core.config as core_config
 from apps.admin.dependencies import create_token, verify_token
+from apps.admin.schemas import LoginData
 from apps.admin.services import ConfigService
+from fastapi import HTTPException
 from main import parse_setup_options
 from core.security import (
     LEGACY_DEFAULT_ADMIN_TOKEN,
@@ -119,6 +123,26 @@ def test_admin_jwt_signature_uses_independent_secret(self):
         with self.assertRaises(ValueError):
             verify_token(token)
 
+    def test_configured_session_lifetime_is_returned_by_login(self):
+        settings.admin_token = hash_password("admin-password")
+        settings.jwt_secret = "j" * 48
+        settings.adminSessionExpire = 90 * 24 * 60 * 60
+        original_time = admin_dependencies.time.time
+        admin_dependencies.time.time = lambda: 1_800_000_000
+        try:
+            response = asyncio.run(
+                admin_views.login(LoginData(password="admin-password"))
+            )
+            payload = verify_token(response.detail["token"])
+        finally:
+            admin_dependencies.time.time = original_time
+
+        self.assertEqual(response.detail["expires_in"], 90 * 24 * 60 * 60)
+        self.assertEqual(
+            response.detail["expires_at"], 1_800_000_000 + 90 * 24 * 60 * 60
+        )
+        self.assertEqual(payload["exp"], response.detail["expires_at"])
+
 
 class FakeKeyValue:
     saved_value = None
@@ -158,6 +182,19 @@ async def update_or_create(cls, key, defaults):
 
 
 class ConfigServiceSecurityTests(SettingsOverrideMixin, unittest.TestCase):
+    def test_admin_session_lifetime_rejects_out_of_range_values(self):
+        settings.user_config = copy.deepcopy(DEFAULT_CONFIG)
+
+        for invalid_value in (0, 86399, 90000, 365 * 24 * 60 * 60 + 1):
+            with self.subTest(invalid_value=invalid_value):
+                with self.assertRaises(HTTPException) as context:
+                    asyncio.run(
+                        ConfigService().update_config(
+                            {"adminSessionExpire": invalid_value}
+                        )
+                    )
+                self.assertEqual(context.exception.status_code, 400)
+
     def test_admin_password_update_rotates_jwt_secret(self):
         old_secret = "j" * 48
         settings.user_config = {