fix: harden share retrieval and admin visibility (#482, #480)

Author: vastsa

apps/base/utils.py

@@ -90,7 +90,7 @@ async def get_expire_info(
 
 
 def get_code_generate_type() -> str:
-    code_generate_type = getattr(settings, "code_generate_type", "number")
+    code_generate_type = getattr(settings, "code_generate_type", "secret")
     if code_generate_type in {"secret", "string"}:
         return "secret"
     return "number"
@@ -127,5 +127,6 @@ async def calculate_file_hash(file: UploadFile, chunk_size=1024 * 1024) -> str:
 
 ip_limit = {
     "error": IPRateLimit(count=settings.errorCount, minutes=settings.errorMinute),
+    "metadata": IPRateLimit(count=settings.errorCount, minutes=settings.errorMinute),
     "upload": IPRateLimit(count=settings.uploadCount, minutes=settings.uploadMinute),
 }

apps/base/views.py

@@ -11,6 +11,7 @@
 from fastapi import APIRouter, Form, Request, UploadFile, File, Depends, HTTPException
 from pydantic import BaseModel, ValidationError
 from starlette import status
+from tortoise.expressions import Case, F, Q, When
 
 from apps.admin.dependencies import share_required_login
 from apps.base.models import FileCodes, UploadChunk, PresignUploadSession
@@ -29,7 +30,12 @@
 from core.response import APIResponse
 from core.settings import settings
 from core.storage import storages, FileStorageInterface
-from core.utils import get_select_token, get_now, sanitize_filename
+from core.utils import (
+    get_file_url as get_proxy_file_url,
+    get_select_token,
+    get_now,
+    sanitize_filename,
+)
 
 share_api = APIRouter(prefix="/share", tags=["分享"])
 
@@ -211,11 +217,25 @@ async def get_code_file_by_code(
     return True, file_code
 
 
-async def update_file_usage(file_code: FileCodes) -> None:
-    file_code.used_count += 1
-    if file_code.expired_count > 0:
-        file_code.expired_count -= 1
-    await file_code.save()
+async def consume_file_usage(file_code: FileCodes) -> bool:
+    """原子校验分享状态并记录一次实际领取。"""
+    now = await get_now()
+    eligible = (
+        Q(expired_count__gt=0)
+        | Q(expired_count__lt=0, expired_at__gt=now)
+        | Q(expired_count__lt=0, expired_at=None)
+    )
+    updated = await FileCodes.filter(Q(id=file_code.id) & eligible).update(
+        expired_count=Case(
+            When(expired_count__gt=0, then=F("expired_count") - 1),
+            default=F("expired_count"),
+        ),
+        used_count=F("used_count") + 1,
+    )
+    if not updated:
+        return False
+    await file_code.refresh_from_db()
+    return True
 
 
 def build_file_metadata(file_code: FileCodes) -> dict:
@@ -242,9 +262,13 @@ async def build_select_detail(
     file_code: FileCodes, file_storage: FileStorageInterface
 ) -> dict:
     metadata = build_file_metadata(file_code)
-    download_url = (
-        None if file_code.text is not None else await file_storage.get_file_url(file_code)
-    )
+    if file_code.text is not None:
+        download_url = None
+    elif file_code.expired_count >= 0:
+        # 有次数限制的文件必须经过下载接口，第三方直链无法阻止重复使用。
+        download_url = await get_proxy_file_url(file_code.code)
+    else:
+        download_url = await file_storage.get_file_url(file_code)
     content = file_code.text if file_code.text is not None else None
     return {
         **metadata,
@@ -255,24 +279,28 @@ async def build_select_detail(
 
 
 @share_api.get("/metadata/")
-async def get_file_metadata(code: str, ip: str = Depends(ip_limit["error"])):
+async def get_file_metadata(code: str, ip: str = Depends(ip_limit["metadata"])):
     has, file_code = await get_code_file_by_code(code)
     if not has:
-        ip_limit["error"].add_ip(ip)
+        ip_limit["metadata"].add_ip(ip)
         return APIResponse(code=404, detail=file_code)
 
     assert isinstance(file_code, FileCodes)
+    ip_limit["metadata"].add_ip(ip)
     return APIResponse(detail=build_file_metadata(file_code))
 
 
 @share_api.post("/metadata/")
-async def post_file_metadata(data: SelectFileModel, ip: str = Depends(ip_limit["error"])):
+async def post_file_metadata(
+    data: SelectFileModel, ip: str = Depends(ip_limit["metadata"])
+):
     has, file_code = await get_code_file_by_code(data.code)
     if not has:
-        ip_limit["error"].add_ip(ip)
+        ip_limit["metadata"].add_ip(ip)
         return APIResponse(code=404, detail=file_code)
 
     assert isinstance(file_code, FileCodes)
+    ip_limit["metadata"].add_ip(ip)
     return APIResponse(detail=build_file_metadata(file_code))
 
 
@@ -285,7 +313,8 @@ async def get_code_file(code: str, ip: str = Depends(ip_limit["error"])):
         return APIResponse(code=404, detail=file_code)
 
     assert isinstance(file_code, FileCodes)
-    await update_file_usage(file_code)
+    if not await consume_file_usage(file_code):
+        return APIResponse(code=404, detail="文件已过期")
     return await file_storage.get_file_response(file_code)
 
 
@@ -298,8 +327,16 @@ async def select_file(data: SelectFileModel, ip: str = Depends(ip_limit["error"]
         return APIResponse(code=404, detail=file_code)
 
     assert isinstance(file_code, FileCodes)
-    await update_file_usage(file_code)
-    return APIResponse(detail=await build_select_detail(file_code, file_storage))
+    detail = await build_select_detail(file_code, file_storage)
+    download_url = detail.get("download_url")
+    consumes_on_download = isinstance(download_url, str) and download_url.startswith(
+        "/share/download?"
+    )
+    if not consumes_on_download and not await consume_file_usage(file_code):
+        return APIResponse(code=404, detail="文件已过期")
+    if not consumes_on_download:
+        detail.update(build_file_metadata(file_code))
+    return APIResponse(detail=detail)
 
 
 @share_api.get("/download")
@@ -309,10 +346,12 @@ async def download_file(key: str, code: str, ip: str = Depends(ip_limit["error"]
     if await get_select_token(normalized_code) != key:
         ip_limit["error"].add_ip(ip)
         raise HTTPException(status_code=403, detail="下载鉴权失败")
-    has, file_code = await get_code_file_by_code(normalized_code, False)
+    has, file_code = await get_code_file_by_code(normalized_code)
     if not has:
-        return APIResponse(code=404, detail="文件不存在")
+        return APIResponse(code=404, detail=file_code)
     assert isinstance(file_code, FileCodes)
+    if not await consume_file_usage(file_code):
+        return APIResponse(code=404, detail="文件已过期")
     return (
         APIResponse(detail=file_code.text)
         if file_code.text

core/config.py

@@ -64,6 +64,8 @@ async def ensure_security_settings() -> None:
 def _sync_ip_limits() -> None:
     ip_limit["error"].minutes = settings.errorMinute
     ip_limit["error"].count = settings.errorCount
+    ip_limit["metadata"].minutes = settings.errorMinute
+    ip_limit["metadata"].count = settings.errorCount
     ip_limit["upload"].minutes = settings.uploadMinute
     ip_limit["upload"].count = settings.uploadCount
 

core/settings.py

@@ -44,7 +44,7 @@
     "uploadSize": 1024 * 1024 * 10,
     "allowed_file_types": ["*"],
     "expireStyle": ["day", "hour", "minute", "forever", "count"],
-    "code_generate_type": "number",
+    "code_generate_type": "secret",
     "uploadMinute": 1,
     "enableChunk": 0,
     "webdav_url": "",

core/tasks.py

@@ -28,6 +28,7 @@ async def delete_expire_files():
                     if not dirs and not files:
                         os.rmdir(root)
             await ip_limit["error"].remove_expired_ip()
+            await ip_limit["metadata"].remove_expired_ip()
             await ip_limit["upload"].remove_expired_ip()
             expire_data = await FileCodes.filter(
                 Q(expired_at__lt=await get_now()) | Q(expired_count=0)

main.py

@@ -33,6 +33,12 @@
 from core.version import APP_VERSION
 
 
+def normalize_public_flag(value) -> int:
+    if isinstance(value, str):
+        return int(value.strip().lower() in {"1", "true", "on", "yes"})
+    return int(bool(value))
+
+
 def build_public_config() -> dict:
     return {
         "name": settings.name,
@@ -45,7 +51,7 @@ def build_public_config() -> dict:
         "openUpload": settings.openUpload,
         "notify_title": settings.notify_title,
         "notify_content": settings.notify_content,
-        "show_admin_address": settings.showAdminAddr,
+        "show_admin_address": normalize_public_flag(settings.showAdminAddr),
         "max_save_seconds": settings.max_save_seconds,
     }
 
@@ -61,7 +67,7 @@ def build_public_meta() -> dict:
         "features": {
             "chunkUpload": bool(settings.enableChunk),
             "guestUpload": bool(settings.openUpload),
-            "adminAddressVisible": bool(settings.showAdminAddr),
+            "adminAddressVisible": bool(normalize_public_flag(settings.showAdminAddr)),
             "expirationModes": settings.expireStyle,
         },
         "limits": {
@@ -153,7 +159,9 @@ def parse_setup_options(data: dict) -> dict:
     if not expire_styles:
         raise ValueError("至少需要选择一种过期方式")
 
-    code_generate_type = get_form_value(data, "code_generate_type", "number")
+    code_generate_type = get_form_value(
+        data, "code_generate_type", DEFAULT_CONFIG["code_generate_type"]
+    )
     if code_generate_type not in {"number", "secret"}:
         raise ValueError("提取码类型不正确")
 
@@ -222,7 +230,9 @@ def build_setup_page(error: str = "", form: dict | None = None) -> str:
     chunk_checked = (
         " checked" if normalize_bool_field(form, "enableChunk", False) else ""
     )
-    code_generate_type = get_form_value(form, "code_generate_type", "number")
+    code_generate_type = get_form_value(
+        form, "code_generate_type", DEFAULT_CONFIG["code_generate_type"]
+    )
     selected_expire_styles = get_form_list(form, "expireStyle") or list(
         DEFAULT_CONFIG["expireStyle"]
     )

tests/test_issue_480_admin_address.py

@@ -0,0 +1,26 @@
+import unittest
+
+from core.settings import settings
+from main import build_public_config, build_public_meta
+
+
+class AdminAddressPublicConfigTests(unittest.TestCase):
+    def setUp(self):
+        self._original_user_config = dict(settings.user_config)
+
+    def tearDown(self):
+        settings.user_config = self._original_user_config
+
+    def test_admin_address_is_exposed_as_strict_binary_flag(self):
+        cases = ((0, 0), (1, 1), ("0", 0), ("1", 1))
+
+        for configured_value, expected_value in cases:
+            with self.subTest(configured_value=configured_value):
+                settings.showAdminAddr = configured_value
+
+                public_value = build_public_config()["show_admin_address"]
+                feature_value = build_public_meta()["features"]["adminAddressVisible"]
+
+                self.assertIs(type(public_value), int)
+                self.assertEqual(public_value, expected_value)
+                self.assertIs(feature_value, bool(expected_value))