管理员工作台
寄件管理
用一枚口令,收齐一份资料。控制有效期、投递次数与收件目录。
+管理员登录
使用现有站点管理员密码。寄件码不能登录此页面。
+ +创建寄件码
+ +
寄件码列表
| 用途 / 编号 | 目标存储 | 有效期 | 状态 | 成功 / 占用 / 总数 | 操作 |
|---|
收到的文件
删除文件不会恢复寄件码的历史成功次数。删除寄件码会撤销投递权限,但保留这里的文件。
+| 文件 | 大小 | 时间 | 状态 | 操作 |
|---|
From 03b17419435da9a8c6e93bd01b7cbb32399e2e3e Mon Sep 17 00:00:00 2001 From: dawnStamp <310845922@qq.com> Date: Tue, 15 Sep 2026 20:01:16 +0800 Subject: [PATCH 1/8] =?UTF-8?q?feat(=E5=AF=84=E4=BB=B6):=20=E6=96=B0?= =?UTF-8?q?=E5=A2=9E=E5=AF=84=E4=BB=B6=E6=8E=88=E6=9D=83=E5=B9=B6=E5=A4=8D?= =?UTF-8?q?=E7=94=A8=E6=99=AE=E9=80=9A=E4=B8=8A=E4=BC=A0=E4=B8=8E=E6=96=87?= =?UTF-8?q?=E4=BB=B6=E7=AE=A1=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/admin/services.py | 21 +- apps/admin/views.py | 2 + apps/base/migrations/migrations_008.py | 42 ++++ apps/base/migrations/migrations_009.py | 11 + apps/base/migrations/migrations_010.py | 11 + apps/base/models.py | 39 +++ apps/base/pages.py | 23 +- apps/base/quota.py | 9 +- apps/base/schemas.py | 14 +- apps/base/services.py | 109 ++++---- apps/base/share_storage.py | 28 +++ apps/base/tasks.py | 16 +- apps/base/upload_access.py | 131 ++++++++++ apps/base/views.py | 126 ++++++---- apps/delivery/__init__.py | 1 + apps/delivery/schemas.py | 71 ++++++ apps/delivery/services.py | 333 +++++++++++++++++++++++++ apps/delivery/static/admin.html | 48 ++++ apps/delivery/static/admin.js | 141 +++++++++++ apps/delivery/static/common.js | 21 ++ apps/delivery/static/delivery.css | 5 + apps/delivery/static/delivery.html | 38 +++ apps/delivery/static/delivery.js | 63 +++++ apps/delivery/static/entry.css | 2 + apps/delivery/static/entry.js | 9 + apps/delivery/static/logo.svg | 2 + apps/delivery/storage.py | 117 +++++++++ apps/delivery/views.py | 218 ++++++++++++++++ core/storage.py | 5 +- docs/.vitepress/config.mts | 4 + docs/guide/delivery.md | 89 +++++++ main.py | 9 + readme.md | 10 +- tests/test_issue_476_theme_assets.py | 5 +- tests/test_security_gaps.py | 5 +- 35 files changed, 1655 insertions(+), 123 deletions(-) create mode 100644 apps/base/migrations/migrations_008.py create mode 100644 apps/base/migrations/migrations_009.py create mode 100644 apps/base/migrations/migrations_010.py create mode 100644 apps/base/share_storage.py create mode 100644 apps/base/upload_access.py create mode 100644 apps/delivery/__init__.py create mode 100644 apps/delivery/schemas.py create mode 100644 apps/delivery/services.py create mode 100644 apps/delivery/static/admin.html create mode 100644 apps/delivery/static/admin.js create mode 100644 apps/delivery/static/common.js create mode 100644 apps/delivery/static/delivery.css create mode 100644 apps/delivery/static/delivery.html create mode 100644 apps/delivery/static/delivery.js create mode 100644 apps/delivery/static/entry.css create mode 100644 apps/delivery/static/entry.js create mode 100644 apps/delivery/static/logo.svg create mode 100644 apps/delivery/storage.py create mode 100644 apps/delivery/views.py create mode 100644 docs/guide/delivery.md diff --git a/apps/admin/services.py b/apps/admin/services.py index ab7a97e11..1c73585ea 100644 --- a/apps/admin/services.py +++ b/apps/admin/services.py @@ -18,7 +18,8 @@ from apps.base.config import refresh_settings from apps.base.services import response_from_download, stored_file_of from core.security import INTERNAL_CONFIG_KEYS, generate_jwt_secret -from apps.base.models import FileCodes, KeyValue +from apps.base.models import DeliveryCode, DeliveryFile, FileCodes, KeyValue +from apps.base.share_storage import remove_delivery_share, storage_for_share from apps.base.utils import get_expire_info, get_file_path_name from apps.base.quota import release_storage, reserve_storage from fastapi import HTTPException @@ -98,6 +99,9 @@ def _file_metadata_key(self, file_id: int) -> str: return f"{self.FILE_METADATA_KEY_PREFIX}{file_id}" async def _delete_file_code(self, file_code: FileCodes): + # 寄件分享在两个管理入口使用相同撤销与清理逻辑,避免重复计费或遗留可用取件码。 + if await remove_delivery_share(file_code): + return if file_code.text is None: await self.file_storage.delete_file(stored_file_of(file_code)) await KeyValue.filter(key=self._file_metadata_key(file_code.id)).delete() @@ -477,6 +481,7 @@ async def list_files( health: str = "", sort_by: str = "created_at", sort_order: str = "desc", + delivery_id: int | None = None, ): page = max(page, 1) size = min(max(size, 1), 100) @@ -487,7 +492,16 @@ async def list_files( sort_by = self._normalize_sort_by(sort_by) reverse = sort_order.strip().lower() != "asc" - all_files = await FileCodes.all() + query = FileCodes.all() + if delivery_id is not None: + # 收件列表复用文件管理的数据与操作,只限定当前管理员选中的寄件码。 + if not await DeliveryCode.filter(id=delivery_id, owner_id="admin").exists(): + raise HTTPException(404, "寄件码不存在") + share_ids = await DeliveryFile.filter( + delivery_id=delivery_id, owner_id="admin", status="shared" + ).values_list("share_id", flat=True) + query = query.filter(id__in=share_ids) + all_files = await query now = await get_now() enriched_files = [] summary = { @@ -1412,7 +1426,8 @@ async def download_file(self, file_id: int): if file_code.text: return APIResponse(detail=file_code.text) else: - return response_from_download(await self.file_storage.get_file_response(stored_file_of(file_code))) + storage = await storage_for_share(file_code, self._file_storage) + return response_from_download(await storage.get_file_response(stored_file_of(file_code))) async def preview_file(self, file_id: int, max_chars: int = 4000): max_chars = min(max(max_chars, 1), 20000) diff --git a/apps/admin/views.py b/apps/admin/views.py index dcbd61ab2..45c9c358b 100644 --- a/apps/admin/views.py +++ b/apps/admin/views.py @@ -381,6 +381,7 @@ async def file_list( health: str = "", sort_by: str = "created_at", sort_order: str = "desc", + delivery_id: Optional[int] = None, file_service: FileService = Depends(get_file_service), ): page = max(page, 1) @@ -394,6 +395,7 @@ async def file_list( health=health, sort_by=sort_by, sort_order=sort_order, + delivery_id=delivery_id, ) return APIResponse( detail={ diff --git a/apps/base/migrations/migrations_008.py b/apps/base/migrations/migrations_008.py new file mode 100644 index 000000000..a3b0701ff --- /dev/null +++ b/apps/base/migrations/migrations_008.py @@ -0,0 +1,42 @@ +"""新增寄件口令与独立收件记录;保留既有分享表及其权限语义。""" + +from tortoise import connections + + +async def migrate(): + # 当前应用使用 SQLite;表定义与 Tortoise 模型保持一致,可重复执行。 + await connections.get("default").execute_script(""" + CREATE TABLE IF NOT EXISTS deliverycode ( + id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, + code_digest VARCHAR(64) NOT NULL UNIQUE, + name VARCHAR(100) NOT NULL, + owner_id VARCHAR(64) NOT NULL DEFAULT 'admin', + storage_type VARCHAR(20) NOT NULL, + target_path VARCHAR(200) NOT NULL, + expires_at TIMESTAMP NOT NULL, + max_uploads INT NOT NULL, + used_count INT NOT NULL DEFAULT 0, + reserved_count INT NOT NULL DEFAULT 0, + enabled INT NOT NULL DEFAULT 1, + deleted INT NOT NULL DEFAULT 0, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + CREATE INDEX IF NOT EXISTS idx_deliverycode_owner ON deliverycode(owner_id); + CREATE TABLE IF NOT EXISTS deliveryfile ( + id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, + delivery_id INT NOT NULL, + owner_id VARCHAR(64) NOT NULL DEFAULT 'admin', + token VARCHAR(64) NOT NULL UNIQUE, + filename VARCHAR(255) NOT NULL DEFAULT '', + stored_name VARCHAR(255) NOT NULL DEFAULT '', + file_path VARCHAR(200) NOT NULL, + storage_type VARCHAR(20) NOT NULL, + size BIGINT NOT NULL DEFAULT 0, + status VARCHAR(20) NOT NULL DEFAULT 'pending', + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + CREATE INDEX IF NOT EXISTS idx_deliveryfile_code ON deliveryfile(delivery_id); + CREATE INDEX IF NOT EXISTS idx_deliveryfile_owner ON deliveryfile(owner_id); + CREATE INDEX IF NOT EXISTS idx_deliveryfile_status ON deliveryfile(status); + """) diff --git a/apps/base/migrations/migrations_009.py b/apps/base/migrations/migrations_009.py new file mode 100644 index 000000000..da296a2ca --- /dev/null +++ b/apps/base/migrations/migrations_009.py @@ -0,0 +1,11 @@ +"""保留管理员可查看的寄件码原文,既有摘要及口令有效性保持不变。""" + +from tortoise import connections + + +async def migrate(): + conn = connections.get("default") + # 可重复执行;旧数据保持 NULL,不能伪造或替换用户此前分发的口令。 + columns = await conn.execute_query_dict("PRAGMA table_info(deliverycode)") + if not any(column["name"] == "code_value" for column in columns): + await conn.execute_script("ALTER TABLE deliverycode ADD COLUMN code_value VARCHAR(64) NULL;") diff --git a/apps/base/migrations/migrations_010.py b/apps/base/migrations/migrations_010.py new file mode 100644 index 000000000..650fef38a --- /dev/null +++ b/apps/base/migrations/migrations_010.py @@ -0,0 +1,11 @@ +"""将新的寄件授权上传关联到普通取件记录,保留历史私有收件。""" + +from tortoise import connections + + +async def migrate(): + conn = connections.get("default") + columns = await conn.execute_query_dict("PRAGMA table_info(deliveryfile)") + if not any(column["name"] == "share_id" for column in columns): + await conn.execute_script("ALTER TABLE deliveryfile ADD COLUMN share_id INT NULL;") + await conn.execute_script("CREATE INDEX IF NOT EXISTS idx_deliveryfile_share_id ON deliveryfile(share_id);") diff --git a/apps/base/models.py b/apps/base/models.py index 3820dee1b..4b2e7f147 100644 --- a/apps/base/models.py +++ b/apps/base/models.py @@ -95,6 +95,45 @@ class StorageReservation(models.Model): expires_at = fields.DatetimeField(index=True) +class DeliveryCode(models.Model): + """只授予投递权限的口令;不进入公开取件码表,避免形成下载授权。""" + + id = fields.IntField(pk=True) + code_digest = fields.CharField(max_length=64, unique=True) + # 与普通取件码一样保留原文供管理员管理;旧记录为 NULL,不能从摘要反推。 + code_value = fields.CharField(max_length=64, null=True) + name = fields.CharField(max_length=100) + owner_id = fields.CharField(max_length=64, default="admin", index=True) + storage_type = fields.CharField(max_length=20) + target_path = fields.CharField(max_length=200) + expires_at = fields.DatetimeField() + max_uploads = fields.IntField() + used_count = fields.IntField(default=0) + reserved_count = fields.IntField(default=0) + enabled = fields.BooleanField(default=True) + deleted = fields.BooleanField(default=False) + created_at = fields.DatetimeField(auto_now_add=True) + + +class DeliveryFile(models.Model): + """寄件收件记录;pending 占用次数,stored 计入永久容量,独立于公开分享。""" + + id = fields.IntField(pk=True) + delivery_id = fields.IntField(index=True) + # 新的授权分享关联普通取件记录;NULL 表示旧版私有收件,绝不自动公开。 + share_id = fields.IntField(null=True, index=True) + owner_id = fields.CharField(max_length=64, default="admin", index=True) + token = fields.CharField(max_length=64, unique=True) + filename = fields.CharField(max_length=255, default="") + stored_name = fields.CharField(max_length=255, default="") + file_path = fields.CharField(max_length=200) + storage_type = fields.CharField(max_length=20) + size = fields.BigIntField(default=0) + status = fields.CharField(max_length=20, default="pending", index=True) + created_at = fields.DatetimeField(auto_now_add=True) + updated_at = fields.DatetimeField(auto_now=True) + + file_codes_pydantic = pydantic_model_creator(FileCodes, name="FileCodes") upload_chunk_pydantic = pydantic_model_creator(UploadChunk, name="UploadChunk") key_value_pydantic = pydantic_model_creator(KeyValue, name="KeyValue") diff --git a/apps/base/pages.py b/apps/base/pages.py index 5bd174324..9c2e67b90 100644 --- a/apps/base/pages.py +++ b/apps/base/pages.py @@ -3,7 +3,7 @@ import html from fastapi import APIRouter, HTTPException, Request -from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse +from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse from apps.base.config import initialize_system, is_runtime_initialized from apps.base.setup_wizard import ( @@ -90,12 +90,19 @@ async def theme_asset(asset_path: str): @router.get("/") async def index(request=None, exc=None): + # 新增寄件接口的 404 保持 JSON 状态码,不能被主题 SPA 回退改成 HTML 200。 + if request is not None and request.url.path.startswith(("/api/delivery", "/admin/delivery", "/delivery-assets/")): + return JSONResponse(status_code=404, content={"detail": getattr(exc, "detail", "资源不存在")}) # Site config is admin input (and during the setup window anyone can claim it); # always escape before injecting into the theme template to prevent stored XSS # (mirrors the setup page). + template = resolve_theme_file("index.html").read_text(encoding="utf-8") + # 原生 2024 寄件页面自行提供导航,旧主题仍保留独立入口以兼容滚动更新。 + if not theme_has_delivery_ui(template): + template = template.replace("
+管理员工作台
用一枚口令,收齐一份资料。控制有效期、投递次数与收件目录。
+使用现有站点管理员密码。寄件码不能登录此页面。
+ +
| 用途 / 编号 | 目标存储 | 有效期 | 状态 | 成功 / 占用 / 总数 | 操作 |
|---|
删除文件不会恢复寄件码的历史成功次数。删除寄件码会撤销投递权限,但保留这里的文件。
+| 文件 | 大小 | 时间 | 状态 | 操作 |
|---|
", '' + '') return HTMLResponse( - content=resolve_theme_file("index.html") - .read_text(encoding="utf-8") + content=template .replace("{{title}}", html.escape(str(settings.name))) .replace("{{description}}", html.escape(str(settings.description))) .replace("{{keywords}}", html.escape(str(settings.keywords))) @@ -135,3 +142,13 @@ async def health_check(): "theme": settings.themes_select, } ) + + +def theme_has_delivery_ui(template: str | None = None) -> bool: + """使用构建产物的功能标记兼容新旧前端,不依赖某个发行版本字符串。""" + if template is None: + path = resolve_theme_root() / "index.html" + if not path.is_file(): + return False + template = path.read_text(encoding="utf-8") + return 'name="filecodebox-features" content="delivery"' in template diff --git a/apps/base/quota.py b/apps/base/quota.py index 988ffc796..9ce5d9aea 100644 --- a/apps/base/quota.py +++ b/apps/base/quota.py @@ -4,7 +4,7 @@ from tortoise import connections from tortoise.functions import Sum -from apps.base.models import FileCodes, StorageReservation +from apps.base.models import DeliveryFile, FileCodes, StorageReservation from core.settings import settings from core.utils import get_now @@ -57,11 +57,15 @@ async def get_storage_usage() -> dict[str, int | None]: now = await get_now() # SQL 聚合:此函数在每次上传配额检查时调用,禁止全表拉取(D3) used_rows = await FileCodes.all().annotate(total=Sum("size")).values("total") + # 私有收件与失败残留计入;shared 已由 FileCodes 计算,不能重复计费。 + delivery_rows = await DeliveryFile.filter(status__in=["stored", "cleanup"]).annotate( + total=Sum("size") + ).values("total") reserved_rows = await StorageReservation.filter(expires_at__gt=now).annotate( total=Sum("size") ).values("total") limit = get_storage_limit() - used_bytes = used_rows[0]["total"] or 0 + used_bytes = (used_rows[0]["total"] or 0) + (delivery_rows[0]["total"] or 0) reserved_bytes = reserved_rows[0]["total"] or 0 return { "limit": limit, @@ -95,6 +99,7 @@ async def reserve_storage(token: str, size: int, ttl_seconds: int) -> None: SELECT {ph[0]}, {ph[1]}, {ph[2]} WHERE ( COALESCE((SELECT SUM(size) FROM filecodes), 0) + + COALESCE((SELECT SUM(size) FROM deliveryfile WHERE status IN ('stored', 'cleanup')), 0) + COALESCE((SELECT SUM(size) FROM storagereservation WHERE expires_at > {ph[3]}), 0) + {ph[4]} ) <= {ph[5]} diff --git a/apps/base/schemas.py b/apps/base/schemas.py index f4cf4a405..ecf5b8112 100644 --- a/apps/base/schemas.py +++ b/apps/base/schemas.py @@ -1,4 +1,4 @@ -from pydantic import BaseModel +from pydantic import BaseModel, Field class SelectFileModel(BaseModel): @@ -7,13 +7,15 @@ class SelectFileModel(BaseModel): class InitChunkUploadModel(BaseModel): file_name: str - chunk_size: int = 5 * 1024 * 1024 - file_size: int + # 在计算分片数之前限制非法或过大的分片,避免除零和单片内存失控。 + chunk_size: int = Field(default=5 * 1024 * 1024, ge=1, le=5 * 1024 * 1024) + file_size: int = Field(ge=1) file_hash: str class CompleteUploadModel(BaseModel): - expire_value: int + # 分享完成沿用正数期限,不能通过分片接口绕开表单约束。 + expire_value: int = Field(ge=1) expire_style: str @@ -21,8 +23,8 @@ class CompleteUploadModel(BaseModel): class PresignUploadInitRequest(BaseModel): """预签名上传初始化请求""" file_name: str - file_size: int - expire_value: int = 1 + file_size: int = Field(ge=1) + expire_value: int = Field(default=1, ge=1) expire_style: str = "day" diff --git a/apps/base/services.py b/apps/base/services.py index 095c46935..e384fa70a 100644 --- a/apps/base/services.py +++ b/apps/base/services.py @@ -12,8 +12,9 @@ from core.logger import logger from core.settings import settings -from core.storage import FileStorageInterface, StoredDownload, StoredFile, storages +from core.storage import FileStorageInterface, StoredDownload, StoredFile +from apps.base.upload_access import prepare_upload, upload_storage, create_upload_share, abort_access from apps.base.file_validation import validate_upload_file from apps.base.models import FileCodes, PresignUploadSession, UploadChunk from apps.base.quota import release_storage, reserve_storage @@ -87,10 +88,6 @@ def chunk_reservation_ttl() -> int: class FileUploadService: """统一的文件上传服务""" - @staticmethod - def _storage() -> FileStorageInterface: - return storages[settings.file_storage]() - @staticmethod async def generate_file_path( file_name: str, upload_id: str | None = None @@ -105,6 +102,7 @@ async def create_file_record( file_path: str, expire_value: int, expire_style: str, + access=None, **extra_fields, ) -> str: """统一创建FileCodes记录,返回code""" @@ -113,7 +111,7 @@ async def create_file_record( ) prefix, suffix = os.path.splitext(file_name) - await FileCodes.create( + await create_upload_share(access, code=code, prefix=prefix, suffix=suffix, @@ -129,17 +127,22 @@ async def create_file_record( @staticmethod async def create_text_share( - text: str, expire_value: int, expire_style: str + text: str, expire_value: int, expire_style: str, access=None ) -> str: """文本分享:配额预留 → 建分享记录 → 释放配额。""" text_size = len(text.encode("utf-8")) + # 文本与普通发送一致存入取件表,并占用一次寄件授权。 + await prepare_upload(access, "Text", text_size, uuid.uuid4().hex) + if access is not None and access.record is not None: + access.record.stored_name = "" + await access.record.save(update_fields=["stored_name"]) token = f"text:{uuid.uuid4().hex}" await reserve_storage(token, text_size, ttl_seconds=300) try: expired_at, expired_count, used_count, code = await get_expire_info( expire_value, expire_style ) - await FileCodes.create( + await create_upload_share(access, code=code, text=text, expired_at=expired_at, @@ -150,25 +153,31 @@ async def create_text_share( ) finally: await release_storage(token) + await abort_access(access) return code @staticmethod async def create_file_share( - file: UploadFile, *, size: int, expire_value: int, expire_style: str + file: UploadFile, *, size: int, expire_value: int, expire_style: str, access=None ) -> dict[str, str]: """文件分享:路径生成 → 配额预留 → 存储写入 → 建分享记录,失败回滚已存文件。""" path, suffix, prefix, uuid_file_name, save_path = ( await FileUploadService.generate_file_path(file.filename or "") ) + # 寄件授权只覆盖路径与归属,保留普通上传校验和存储流程。 + _, delivery_path = await prepare_upload(access, file.filename, size, uuid.uuid4().hex) + if delivery_path: + save_path = delivery_path + path, uuid_file_name = os.path.split(save_path) token = f"file:{uuid.uuid4().hex}" await reserve_storage(token, size, ttl_seconds=3600) - storage = FileUploadService._storage() + storage = await upload_storage(access) try: expired_at, expired_count, used_count, code = await get_expire_info( expire_value, expire_style ) await storage.save_file(file.file, save_path, file.content_type) - await FileCodes.create( + await create_upload_share(access, code=code, prefix=prefix, suffix=suffix, @@ -186,18 +195,19 @@ async def create_file_share( raise finally: await release_storage(token) + await abort_access(access) return {"code": code, "name": file.filename} @staticmethod async def complete_chunked_upload( - upload_id: str, chunk_info: UploadChunk, expire_value: int, expire_style: str + upload_id: str, chunk_info: UploadChunk, expire_value: int, expire_style: str, access=None ) -> dict[str, str]: """分片合并:配额 → 完整性/大小校验 → 合并 → 建分享记录 → 清理分片。 失败路径的配额释放与清理范围与原实现逐一对齐: 完整性校验失败仅抛 400(预留由 TTL 兜底);合并失败清理分片文件后抛 500。 """ - storage = FileUploadService._storage() + storage = await upload_storage(access) await reserve_storage( f"chunk:{upload_id}", chunk_info.file_size, ttl_seconds=chunk_reservation_ttl() ) @@ -208,8 +218,8 @@ async def complete_chunked_upload( if len(completed_chunks) != chunk_info.total_chunks: raise HTTPException(400, "分片不完整") - # 用分片数 * chunk_size 校验最大可能大小 - max_total_size = len(completed_chunks) * chunk_info.chunk_size + # 每片已按声明范围校验,合并时使用实际文件大小,避免将尾片向上取整。 + max_total_size = chunk_info.file_size if max_total_size > settings.upload_size: save_path = chunk_info.save_path if save_path: @@ -245,7 +255,7 @@ async def complete_chunked_upload( expired_at, expired_count, used_count, code = await get_expire_info( expire_value, expire_style ) - await FileCodes.create( + await create_upload_share(access, code=code, file_hash=file_hash, # 使用合并后计算的哈希 is_chunked=True, @@ -259,10 +269,14 @@ async def complete_chunked_upload( prefix=prefix, suffix=suffix, ) - await storage.clean_chunks(upload_id, save_path) - await UploadChunk.filter(upload_id=upload_id).delete() + try: + await storage.clean_chunks(upload_id, save_path) + await UploadChunk.filter(upload_id=upload_id).delete() + except Exception: + logger.warning("分享已创建,分片清理稍后重试 upload_id=%s", upload_id, exc_info=True) await release_storage(f"chunk:{upload_id}") - return {"code": code, "name": safe_file_name} + # 寄件存储名带唯一前缀,但发送结果仍展示原文件名。 + return {"code": code, "name": access.record.filename if access is not None and access.record is not None else safe_file_name} except ValueError as e: raise HTTPException(400, str(e)) except Exception as e: @@ -279,7 +293,7 @@ async def complete_chunked_upload( @staticmethod async def commit_proxy_upload( - session: PresignUploadSession, file: UploadFile + session: PresignUploadSession, file: UploadFile, access=None ) -> str: """预签名代理上传:配额 → 大小/类型/一致性校验 → 转存 → 建记录 → 会话清理。 @@ -296,36 +310,18 @@ async def commit_proxy_upload( if abs(file_size - session.file_size) > 1024: raise HTTPException(400, "文件大小与声明不符") - storage = FileUploadService._storage() + storage = await upload_storage(access) try: await storage.save_file(file.file, session.save_path, file.content_type) except Exception as e: raise HTTPException(500, f"文件保存失败: {str(e)}") - try: - code = await FileUploadService.create_file_record( - session.file_name, - file_size, - os.path.dirname(session.save_path), - session.expire_value, - session.expire_style, - ) - except Exception: - await rollback_saved_file( - storage, - os.path.dirname(session.save_path), - os.path.basename(session.save_path), - context="预签名代理上传:记录创建失败", - upload_id=session.upload_id, - ) - raise - - await session.delete() - await release_storage(f"presign:{session.upload_id}") - return code + return await FileUploadService._commit_presign_record( + session, file_size, storage, access=access, context="预签名代理上传:记录创建失败" + ) @staticmethod - async def confirm_direct_upload(session: PresignUploadSession) -> str: + async def confirm_direct_upload(session: PresignUploadSession, access=None) -> str: """预签名直传确认:配额 → 文件存在性 → 建记录 → 会话清理。 预留失败说明配额已耗尽,此时清理远端临时文件与会话后原样抛出。 @@ -337,7 +333,7 @@ async def confirm_direct_upload(session: PresignUploadSession) -> str: ttl_seconds=PRESIGN_SESSION_EXPIRES, ) except HTTPException: - storage = FileUploadService._storage() + storage = await upload_storage(access) try: if await storage.file_exists(session.save_path): await storage.delete_file( @@ -351,33 +347,34 @@ async def confirm_direct_upload(session: PresignUploadSession) -> str: await release_storage(f"presign:{session.upload_id}") raise - storage = FileUploadService._storage() + storage = await upload_storage(access) if not await storage.file_exists(session.save_path): raise HTTPException(404, "文件未上传或上传失败") + return await FileUploadService._commit_presign_record( + session, session.file_size, storage, access=access, context="预签名确认:记录创建失败" + ) + + @staticmethod + async def _commit_presign_record(session, file_size, storage, *, access=None, context): + """代理上传与直传共用记录提交、失败回滚及会话释放,避免两条路径行为分叉。""" try: code = await FileUploadService.create_file_record( - session.file_name, - session.file_size, - os.path.dirname(session.save_path), - session.expire_value, - session.expire_style, + session.file_name, file_size, os.path.dirname(session.save_path), + session.expire_value, session.expire_style, access=access, ) except Exception: await rollback_saved_file( - storage, - os.path.dirname(session.save_path), - os.path.basename(session.save_path), - context="预签名确认:记录创建失败", - upload_id=session.upload_id, + storage, os.path.dirname(session.save_path), os.path.basename(session.save_path), + context=context, upload_id=session.upload_id, ) raise - await session.delete() await release_storage(f"presign:{session.upload_id}") return code + def response_from_download(download: StoredDownload): """Build the starlette Response for a StoredDownload (view-layer duty).""" if download.path is not None: diff --git a/apps/base/share_storage.py b/apps/base/share_storage.py new file mode 100644 index 000000000..fb30c713b --- /dev/null +++ b/apps/base/share_storage.py @@ -0,0 +1,28 @@ +"""寄件生成的普通分享使用原投递后端,避免全站存储切换后找错文件。""" + +from apps.base.models import DeliveryFile +from core.settings import settings +from core.storage import storages + + +async def delivery_record(file_code): + return await DeliveryFile.filter(share_id=file_code.id).first() + + +async def storage_for_share(file_code, fallback=None): + record = await delivery_record(file_code) + if record is not None: + # 延迟导入以保持应用模块边界,沿用 OneDrive 的精确对象键适配。 + from apps.delivery.storage import get_storage + return await get_storage(record.storage_type) + return fallback if fallback is not None else storages[settings.file_storage]() + + +async def remove_delivery_share(file_code): + """先撤销取件记录再清理文件;失败残留继续占用容量并交由后台重试。""" + record = await delivery_record(file_code) + if record is None: + return False + from apps.delivery.services import request_file_removal + await request_file_removal(record.id) + return True diff --git a/apps/base/tasks.py b/apps/base/tasks.py index 4a682877e..ad2fcf374 100644 --- a/apps/base/tasks.py +++ b/apps/base/tasks.py @@ -20,6 +20,7 @@ from core.settings import settings, data_root from core.storage import FileStorageInterface, StoredFile, storages from apps.base.services import stored_file_of +from apps.base.share_storage import remove_delivery_share from core.utils import get_now @@ -27,7 +28,7 @@ async def delete_expire_files(): while True: try: await refresh_settings() - file_storage: FileStorageInterface = storages[settings.file_storage]() + file_storage = None # 遍历 share目录下的所有文件夹,删除空的文件夹,并判断父目录是否为空,如果为空也删除 if settings.file_storage == "local": for root, dirs, files in os.walk(f"{data_root}/share/data"): @@ -42,6 +43,15 @@ async def delete_expire_files(): ).all() for exp in expire_data: try: + if await remove_delivery_share(exp): + continue + except Exception: + # 关联清理事务失败时保留分享,下一轮重试,不能落入普通删除分支丢失关联。 + logger.warning("寄件分享过期清理失败 id=%s", exp.id, exc_info=True) + continue + try: + if file_storage is None: + file_storage = storages[settings.file_storage]() await file_storage.delete_file(stored_file_of(exp)) except Exception as e: logger.error(f"删除过期文件失败 code={exp.code}: {e}") @@ -65,7 +75,7 @@ async def clean_incomplete_uploads(): expire_time = now - datetime.timedelta(hours=expire_hours) expired_sessions = await UploadChunk.filter( chunk_index=-1, created_at__lt=expire_time - ).all() + ).exclude(upload_id__startswith="d_").all() for session in expired_sessions: try: @@ -105,7 +115,7 @@ async def clean_expired_presign_sessions(): now = await get_now() expired_sessions = await PresignUploadSession.filter( expires_at__lt=now - ).all() + ).exclude(upload_id__startswith="d_").all() for session in expired_sessions: if session.mode == "direct": try: diff --git a/apps/base/upload_access.py b/apps/base/upload_access.py new file mode 100644 index 000000000..4030b2d92 --- /dev/null +++ b/apps/base/upload_access.py @@ -0,0 +1,131 @@ +"""普通上传的可选寄件授权层;通过显式参数传递,不修改全站配置或管理员会话。""" + +import asyncio +from dataclasses import dataclass + +from fastapi import Header, HTTPException, Request + +from apps.base.models import DeliveryFile, FileCodes +from core.settings import settings +from core.storage import storages +from core.utils import get_now + + +@dataclass +class UploadAccess: + """每个请求独立的上传身份;寄件记录的 token 同时绑定普通上传会话。""" + + code_id: int | None = None + record: DeliveryFile | None = None + + +async def authorize_upload(request: Request, authorization: str | None = Header(default=None)): + """校验每一步的授权和会话归属,游客模式也不能访问寄件上传会话。""" + from apps.admin.dependencies import share_required_login, verify_token + from apps.delivery.services import active_code, heartbeat + + access = UploadAccess() + if authorization and authorization.startswith("Bearer "): + try: + payload = verify_token(authorization[7:]) + except ValueError: + raise HTTPException(401, "上传凭证无效或已过期") from None + if payload.get("purpose") == "delivery" and not payload.get("is_admin"): + access.code_id = int(payload["delivery_id"]) + await active_code(access.code_id) + if access.code_id is None: + await share_required_login(authorization) + + upload_id = request.path_params.get("upload_id") + if upload_id: + record = await DeliveryFile.filter(token=upload_id).first() + if record is not None or upload_id.startswith("d_"): + if record is None or record.delivery_id != access.code_id: + raise HTTPException(404, "上传会话不存在") + if record.status not in {"pending", "shared"}: + raise HTTPException(409, "上传会话正在完成或清理,请稍后重试") + access.record = record + elif access.code_id is not None: + raise HTTPException(404, "上传会话不属于该寄件码") + + # 完成阶段原子抢占,避免并发合并或代理上传覆盖同一个已发布文件。 + finalizing = False + if access.record is not None and access.record.status == "pending" and ( + "/complete/" in request.url.path or "/confirm/" in request.url.path or "/proxy/" in request.url.path + ): + changed = await DeliveryFile.filter(id=access.record.id, status="pending").update(status="finalizing", updated_at=await get_now()) + if changed != 1: + raise HTTPException(409, "上传正在完成,请稍后重试") + access.record.status = "finalizing" + finalizing = True + # 长时间传输续租;新会话在初始化后由后续分片请求续租,空闲会话由后台回收。 + task = None + if access.record is not None and access.record.status in {"pending", "finalizing"}: + await DeliveryFile.filter(id=access.record.id).update(updated_at=await get_now()) + task = asyncio.create_task(heartbeat(access.record)) + try: + yield access + except BaseException: + # 初始化、文本或单文件上传失败即释放本次占用;分片请求失败则保留已传内容供续传。 + if not upload_id and access.record is not None: + await asyncio.shield(abort_access(access)) + raise + finally: + if finalizing: + # 失败仍可重新提交完成请求;成功记录已变为 shared,不会被回退。 + await DeliveryFile.filter(id=access.record.id, status="finalizing").update(status="pending", updated_at=await get_now()) + if task: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +async def prepare_upload(access, file_name, file_size, upload_id): + """仅寄件上传预占一次投递次数,并将目标存储和路径固定到上传会话。""" + if access is None or access.code_id is None: + return upload_id, None + from apps.delivery.services import reserve_slot, normalize_delivery_filename + + record = await reserve_slot(access.code_id) + # 先绑定记录,后续文件名处理或会话保存失败时依赖退出逻辑仍能释放占用。 + access.record = record + name = await normalize_delivery_filename(file_name) + record.token = "d_" + upload_id + record.filename = name + record.stored_name = record.token + "_" + name + record.size = file_size + await record.save() + return record.token, f"{record.file_path}/{record.stored_name}" + + +async def upload_storage(access=None): + """上传和合并均使用寄件码选定的存储,普通上传仍遵循全站设置。""" + if access is not None and access.record is not None: + from apps.delivery.storage import get_storage + return await get_storage(access.record.storage_type) + return storages[settings.file_storage]() + + +async def create_upload_share(access=None, **fields): + """共用普通取件记录;寄件扣次与关联记录提交必须处于同一事务。""" + if access is None or access.record is None: + return await FileCodes.create(**fields) + # 授权适配层只传递身份,扣次及私有/公开收件事务集中在寄件业务层。 + from apps.delivery.services import commit_delivery + return await commit_delivery(access.record, fields) + + +async def abort_access(access): + """取消或初始化失败释放寄件占用,实际残留交由原清理流程重试。""" + if access is not None and access.record is not None: + from apps.delivery.services import abort_upload + await abort_upload(access.record.id) + + +async def completed_upload(access): + """完成响应丢失后允许按原会话取回结果,不再合并、覆盖文件或重复扣次。""" + if access is None or access.record is None or access.record.status != "shared": + return None + share = await FileCodes.filter(id=access.record.share_id).first() + if share is None or await share.is_expired(): + raise HTTPException(410, "该上传的文件已过期") + return {"code": share.code, "name": access.record.filename} diff --git a/apps/base/views.py b/apps/base/views.py index ef54a84b4..e176a372c 100644 --- a/apps/base/views.py +++ b/apps/base/views.py @@ -4,7 +4,7 @@ from datetime import timedelta from urllib.parse import quote, unquote -from typing import Optional, Tuple, Union +from typing import Annotated, Optional, Tuple, Union from fastapi import APIRouter, Form, Request, UploadFile, File, Depends, HTTPException from pydantic import BaseModel, ValidationError @@ -12,9 +12,11 @@ from starlette.responses import Response from tortoise.expressions import Case, F, Q, When -from apps.admin.dependencies import share_required_login +from apps.base.upload_access import UploadAccess, authorize_upload, prepare_upload, upload_storage, abort_access, completed_upload +from apps.base.models import DeliveryFile from apps.base.models import FileCodes, UploadChunk, PresignUploadSession from apps.base.quota import release_storage, reserve_storage +from apps.base.share_storage import delivery_record, storage_for_share from apps.base.services import ( PRESIGN_SESSION_EXPIRES, FileUploadService, @@ -38,7 +40,7 @@ ) from core.response import APIResponse from core.settings import settings -from core.storage import storages, FileStorageInterface +from core.storage import FileStorageInterface, storages as storages from core.utils import ( get_file_url as get_proxy_file_url, get_select_token, @@ -56,8 +58,9 @@ def normalize_share_code(code: str) -> str: return str(code or "").strip() -@share_api.post("/text/", dependencies=[Depends(share_required_login)]) +@share_api.post("/text/", dependencies=[Depends(authorize_upload)]) async def share_text( + access: Annotated[UploadAccess, Depends(authorize_upload)] = None, text: str = Form(...), expire_value: int = Form(default=1, gt=0), expire_style: str = Form(default="day"), @@ -69,13 +72,14 @@ async def share_text( if text_size > max_txt_size: raise HTTPException(status_code=403, detail="内容过多,建议采用文件形式") - code = await FileUploadService.create_text_share(text, expire_value, expire_style) + code = await FileUploadService.create_text_share(text, expire_value, expire_style, access=access) ip_limit["upload"].add_ip(ip) return APIResponse(detail={"code": code}) -@share_api.post("/file/", dependencies=[Depends(share_required_login)]) +@share_api.post("/file/", dependencies=[Depends(authorize_upload)]) async def share_file( + access: Annotated[UploadAccess, Depends(authorize_upload)] = None, expire_value: int = Form(default=1, gt=0), expire_style: str = Form(default="day"), file: UploadFile = File(...), @@ -85,7 +89,7 @@ async def share_file( await validate_upload_file(file) validate_expire_style(expire_style) detail = await FileUploadService.create_file_share( - file, size=file_size, expire_value=expire_value, expire_style=expire_style + file, size=file_size, expire_value=expire_value, expire_style=expire_style, access=access ) ip_limit["upload"].add_ip(ip) return APIResponse(detail=detail) @@ -151,7 +155,7 @@ async def build_select_detail( metadata = build_file_metadata(file_code) if file_code.text is not None: download_url = None - elif file_code.expired_count >= 0: + elif file_code.expired_count >= 0 or await delivery_record(file_code) is not None: # 有次数限制的文件必须经过下载接口,第三方直链无法阻止重复使用。 download_url = await get_proxy_file_url(file_code.code) else: @@ -193,7 +197,6 @@ async def post_file_metadata( @share_api.get("/select/") async def get_code_file(code: str, ip: str = Depends(ip_limit["error"])): - file_storage: FileStorageInterface = storages[settings.file_storage]() has, file_code = await get_code_file_by_code(code) if not has: ip_limit["error"].add_ip(ip) @@ -213,18 +216,19 @@ async def get_code_file(code: str, ip: str = Depends(ip_limit["error"])): ) }, ) + file_storage = await storage_for_share(file_code) return response_from_download(await file_storage.get_file_response(stored_file_of(file_code))) @share_api.post("/select/") async def select_file(data: SelectFileModel, ip: str = Depends(ip_limit["error"])): - file_storage: FileStorageInterface = storages[settings.file_storage]() has, file_code = await get_code_file_by_code(data.code) if not has: ip_limit["error"].add_ip(ip) return APIResponse(code=404, detail=file_code) assert isinstance(file_code, FileCodes) + file_storage = await storage_for_share(file_code) 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( @@ -239,7 +243,6 @@ async def select_file(data: SelectFileModel, ip: str = Depends(ip_limit["error"] @share_api.get("/download") async def download_file(key: str, code: str, ip: str = Depends(ip_limit["error"])): - file_storage: FileStorageInterface = storages[settings.file_storage]() normalized_code = normalize_share_code(code) # 同时接受当前窗口与上一窗口 token,避免时间窗边界竞态导致偶发 403 valid_keys = { @@ -255,6 +258,7 @@ async def download_file(key: str, code: str, ip: str = Depends(ip_limit["error"] assert isinstance(file_code, FileCodes) if not await consume_file_usage(file_code): return APIResponse(code=404, detail="文件已过期") + file_storage = await storage_for_share(file_code) return ( APIResponse(detail=file_code.text) if file_code.text @@ -287,21 +291,27 @@ async def parse_complete_upload(request: Request) -> CompleteUploadModel: return await parse_body_model(request, CompleteUploadModel) -@chunk_api.post("/upload/init/", dependencies=[Depends(share_required_login)]) -async def init_chunk_upload(data: InitChunkUploadModel = Depends(parse_init_chunk_upload)): +@chunk_api.post("/upload/init/", dependencies=[Depends(authorize_upload)]) +async def init_chunk_upload(data: InitChunkUploadModel = Depends(parse_init_chunk_upload), access: Annotated[UploadAccess, Depends(authorize_upload)] = None): + # 保持服务函数可被内部调用;HTTP 请求始终由依赖提供校验后的授权。 + access = access or UploadAccess() safe_file_name = await sanitize_filename(unquote(data.file_name or "")) validate_file_type(safe_file_name) - # 服务端校验:根据 total_chunks * chunk_size 计算理论最大上传量 + # 使用文件真实声明大小校验上限,最后一个分片通常小于整片大小。 total_chunks = (data.file_size + data.chunk_size - 1) // data.chunk_size - max_possible_size = total_chunks * data.chunk_size - if max_possible_size > settings.upload_size: + if data.file_size > settings.upload_size: max_size_mb = settings.upload_size / (1024 * 1024) raise HTTPException( status_code=403, detail=f"文件大小超过限制,最大为 {max_size_mb:.2f} MB" ) - # 断点续传:检查是否存在相同文件的未完成上传会话 - existing_session = await UploadChunk.filter( + # 断点续传按寄件码隔离;普通上传不能恢复凭码创建的会话。 + if access.code_id is not None: + tokens = await DeliveryFile.filter(delivery_id=access.code_id, status="pending").values_list("token", flat=True) + session_scope = UploadChunk.filter(upload_id__in=tokens) + else: + session_scope = UploadChunk.exclude(upload_id__startswith="d_") + existing_session = await session_scope.filter( chunk_hash=data.file_hash, chunk_index=-1, file_size=data.file_size, @@ -309,7 +319,10 @@ async def init_chunk_upload(data: InitChunkUploadModel = Depends(parse_init_chun ).first() if existing_session: + if access.code_id is not None: + access.record = await DeliveryFile.get(token=existing_session.upload_id) if not existing_session.save_path: + await abort_access(access) await UploadChunk.filter(upload_id=existing_session.upload_id).delete() await release_storage(f"chunk:{existing_session.upload_id}") else: @@ -336,6 +349,7 @@ async def init_chunk_upload(data: InitChunkUploadModel = Depends(parse_init_chun # 创建新的上传会话 upload_id = uuid.uuid4().hex + upload_id, delivery_path = await prepare_upload(access, safe_file_name, data.file_size, upload_id) reservation_token = f"chunk:{upload_id}" chunk_expire_seconds = max(1, int(getattr(settings, "chunk_expire_hours", 24))) * 3600 await reserve_storage( @@ -353,9 +367,10 @@ async def init_chunk_upload(data: InitChunkUploadModel = Depends(parse_init_chun chunk_size=data.chunk_size, chunk_hash=data.file_hash, file_name=safe_file_name, - save_path=save_path, + save_path=delivery_path or save_path, ) except Exception: + await abort_access(access) await release_storage(reservation_token) raise return APIResponse( @@ -371,13 +386,17 @@ async def init_chunk_upload(data: InitChunkUploadModel = Depends(parse_init_chun @chunk_api.post( "/upload/chunk/{upload_id}/{chunk_index}", - dependencies=[Depends(share_required_login)], + dependencies=[Depends(authorize_upload)], ) async def upload_chunk( upload_id: str, chunk_index: int, + access: Annotated[UploadAccess, Depends(authorize_upload)] = None, chunk: UploadFile = File(...), ): + # 已完成的寄件会话不能继续写分片,避免覆盖正在供下载的文件。 + if access is not None and access.record is not None and access.record.status != "pending": + raise HTTPException(409, "上传已经完成") # 获取上传会话信息 chunk_info = await UploadChunk.filter(upload_id=upload_id, chunk_index=-1).first() if not chunk_info: @@ -406,6 +425,9 @@ async def upload_chunk( if chunk_index == 0: validate_header_bytes(chunk_info.file_name, None, chunk_data[:64]) chunk_size = len(chunk_data) + expected_size = min(chunk_info.chunk_size, chunk_info.file_size - chunk_index * chunk_info.chunk_size) + if chunk_size != expected_size: + raise HTTPException(400, "分片大小与声明的文件范围不一致") # 校验分片大小不超过声明的 chunk_size if chunk_size > chunk_info.chunk_size: @@ -431,7 +453,7 @@ async def upload_chunk( save_path = chunk_info.save_path # 保存分片到存储 - storage = storages[settings.file_storage]() + storage = await upload_storage(access) try: await storage.save_chunk( upload_id, chunk_index, chunk_data, chunk_hash, save_path @@ -458,9 +480,12 @@ async def upload_chunk( return APIResponse(detail={"chunk_hash": chunk_hash}) -@chunk_api.delete("/upload/{upload_id}", dependencies=[Depends(share_required_login)]) -async def cancel_upload(upload_id: str): +@chunk_api.delete("/upload/{upload_id}", dependencies=[Depends(authorize_upload)]) +async def cancel_upload(upload_id: str, access: Annotated[UploadAccess, Depends(authorize_upload)] = None): """取消上传并清理临时文件""" + if access is not None and access.record is not None: + await abort_access(access) + return APIResponse(detail={"message": "上传已取消"}) chunk_info = await UploadChunk.filter(upload_id=upload_id, chunk_index=-1).first() if not chunk_info: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="上传会话不存在") @@ -468,7 +493,7 @@ async def cancel_upload(upload_id: str): save_path = chunk_info.save_path # 清理存储中的临时文件 - storage = storages[settings.file_storage]() + storage = await upload_storage(access) if save_path: try: await storage.clean_chunks(upload_id, save_path) @@ -483,9 +508,9 @@ async def cancel_upload(upload_id: str): @chunk_api.get( - "/upload/status/{upload_id}", dependencies=[Depends(share_required_login)] + "/upload/status/{upload_id}", dependencies=[Depends(authorize_upload)] ) -async def get_upload_status(upload_id: str): +async def get_upload_status(upload_id: str, access: Annotated[UploadAccess, Depends(authorize_upload)] = None): """获取上传状态""" chunk_info = await UploadChunk.filter(upload_id=upload_id, chunk_index=-1).first() if not chunk_info: @@ -510,20 +535,24 @@ async def get_upload_status(upload_id: str): @chunk_api.post( - "/upload/complete/{upload_id}", dependencies=[Depends(share_required_login)] + "/upload/complete/{upload_id}", dependencies=[Depends(authorize_upload)] ) async def complete_upload( upload_id: str, + access: Annotated[UploadAccess, Depends(authorize_upload)] = None, data: CompleteUploadModel = Depends(parse_complete_upload), ip: str = Depends(ip_limit["upload"]), ): + result = await completed_upload(access) + if result: + return APIResponse(detail=result) # 获取上传基本信息 chunk_info = await UploadChunk.filter(upload_id=upload_id, chunk_index=-1).first() if not chunk_info: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="上传会话不存在") validate_expire_style(data.expire_style) detail = await FileUploadService.complete_chunked_upload( - upload_id, chunk_info, data.expire_value, data.expire_style + upload_id, chunk_info, data.expire_value, data.expire_style, access=access ) ip_limit["upload"].add_ip(ip) return APIResponse(detail=detail) @@ -557,9 +586,9 @@ async def _get_valid_session( return session -@presign_api.post("/upload/init", dependencies=[Depends(share_required_login)]) +@presign_api.post("/upload/init", dependencies=[Depends(authorize_upload)]) async def presign_upload_init( - data: PresignUploadInitRequest, ip: str = Depends(ip_limit["upload"]) + data: PresignUploadInitRequest, access: Annotated[UploadAccess, Depends(authorize_upload)] = None, ip: str = Depends(ip_limit["upload"]) ): """初始化预签名上传,S3返回直传URL,其他存储返回代理URL""" validate_file_type(data.file_name) @@ -571,6 +600,7 @@ async def presign_upload_init( validate_expire_style(data.expire_style) upload_id = uuid.uuid4().hex + upload_id, delivery_path = await prepare_upload(access, data.file_name, data.file_size, upload_id) reservation_token = f"presign:{upload_id}" await reserve_storage( reservation_token, data.file_size, ttl_seconds=PRESIGN_SESSION_EXPIRES @@ -579,7 +609,9 @@ async def presign_upload_init( path, _, _, filename, save_path = await FileUploadService.generate_file_path( data.file_name, upload_id ) - storage: FileStorageInterface = storages[settings.file_storage]() + if delivery_path: + save_path = delivery_path + storage: FileStorageInterface = await upload_storage(access) presigned_url = await storage.generate_presigned_upload_url( save_path, PRESIGN_SESSION_EXPIRES ) @@ -597,6 +629,7 @@ async def presign_upload_init( expires_at=await get_now() + timedelta(seconds=PRESIGN_SESSION_EXPIRES), ) except Exception: + await abort_access(access) await release_storage(reservation_token) raise @@ -616,33 +649,39 @@ async def presign_upload_init( @presign_api.put( - "/upload/proxy/{upload_id}", dependencies=[Depends(share_required_login)] + "/upload/proxy/{upload_id}", dependencies=[Depends(authorize_upload)] ) async def presign_upload_proxy( - upload_id: str, file: UploadFile = File(...), ip: str = Depends(ip_limit["upload"]) + upload_id: str, file: UploadFile = File(...), access: Annotated[UploadAccess, Depends(authorize_upload)] = None, ip: str = Depends(ip_limit["upload"]) ): """代理模式上传,服务器转存到存储后端""" + result = await completed_upload(access) + if result: + return APIResponse(detail=result) session = await _get_valid_session(upload_id, expected_mode="proxy") - code = await FileUploadService.commit_proxy_upload(session, file) + code = await FileUploadService.commit_proxy_upload(session, file, access=access) ip_limit["upload"].add_ip(ip) return APIResponse(detail={"code": code, "name": session.file_name}) @presign_api.post( - "/upload/confirm/{upload_id}", dependencies=[Depends(share_required_login)] + "/upload/confirm/{upload_id}", dependencies=[Depends(authorize_upload)] ) -async def presign_upload_confirm(upload_id: str, ip: str = Depends(ip_limit["upload"])): +async def presign_upload_confirm(upload_id: str, access: Annotated[UploadAccess, Depends(authorize_upload)] = None, ip: str = Depends(ip_limit["upload"])): """直传确认,客户端完成S3直传后调用获取分享码""" + result = await completed_upload(access) + if result: + return APIResponse(detail=result) session = await _get_valid_session(upload_id, expected_mode="direct") - code = await FileUploadService.confirm_direct_upload(session) + code = await FileUploadService.confirm_direct_upload(session, access=access) ip_limit["upload"].add_ip(ip) return APIResponse(detail={"code": code, "name": session.file_name}) @presign_api.get( - "/upload/status/{upload_id}", dependencies=[Depends(share_required_login)] + "/upload/status/{upload_id}", dependencies=[Depends(authorize_upload)] ) -async def presign_upload_status(upload_id: str): +async def presign_upload_status(upload_id: str, access: Annotated[UploadAccess, Depends(authorize_upload)] = None): """查询上传会话状态""" session = await PresignUploadSession.filter(upload_id=upload_id).first() if not session: @@ -661,15 +700,18 @@ async def presign_upload_status(upload_id: str): ) -@presign_api.delete("/upload/{upload_id}", dependencies=[Depends(share_required_login)]) -async def presign_upload_cancel(upload_id: str): +@presign_api.delete("/upload/{upload_id}", dependencies=[Depends(authorize_upload)]) +async def presign_upload_cancel(upload_id: str, access: Annotated[UploadAccess, Depends(authorize_upload)] = None): """取消上传会话""" + if access is not None and access.record is not None: + await abort_access(access) + return APIResponse(detail={"message": "上传会话已取消"}) session = await PresignUploadSession.filter(upload_id=upload_id).first() if not session: raise HTTPException(404, "上传会话不存在") if session.mode == "direct": - storage: FileStorageInterface = storages[settings.file_storage]() + storage: FileStorageInterface = await upload_storage(access) try: if await storage.file_exists(session.save_path): temp_file_code = StoredFile( diff --git a/apps/delivery/__init__.py b/apps/delivery/__init__.py new file mode 100644 index 000000000..36e9f5d55 --- /dev/null +++ b/apps/delivery/__init__.py @@ -0,0 +1 @@ +"""文件驿站寄件模块:独立上传权限、收件管理及内置页面。""" diff --git a/apps/delivery/schemas.py b/apps/delivery/schemas.py new file mode 100644 index 000000000..0c628a1b1 --- /dev/null +++ b/apps/delivery/schemas.py @@ -0,0 +1,71 @@ +"""寄件参数校验,所有路径都相对于已配置的存储根目录。""" + +import re +from datetime import datetime, timezone, timedelta +from pathlib import PurePosixPath + +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +class CreateDeliveryCode(BaseModel): + # 禁止静默接受 owner_id 等越权字段,未来多用户必须由服务端身份决定归属。 + model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) + name: str = Field(min_length=1, max_length=100) + code: str = Field(default="", max_length=64) + storage_type: str = "local" + target_path: str = Field(min_length=1, max_length=200) + expires_at: datetime + max_uploads: int = Field(default=1, ge=1, le=100000) + + @field_validator("code") + @classmethod + def validate_code(cls, value): + if value and not re.fullmatch(r"[A-Za-z0-9_-]{8,64}", value): + raise ValueError("寄件码须为 8 至 64 位字母、数字、下划线或短横线") + return value + + @field_validator("storage_type") + @classmethod + def validate_storage(cls, value): + if value not in {"local", "s3", "webdav", "onedrive", "opendal"}: + raise ValueError("不支持的存储类型") + return value + + @field_validator("target_path") + @classmethod + def validate_path(cls, value): + # 同时约束 POSIX、Windows 与 URL 语义,避免 WebDAV 二次解码或盘符逃逸。 + parts = value.split("/") + if (PurePosixPath(value).is_absolute() + or any(part in {"", ".", ".."} for part in parts) + or not re.fullmatch(r"[\w ./-]+", value, re.UNICODE) + or any(part.endswith((".", " ")) for part in parts) + or any(re.fullmatch(r"(?i)(con|prn|aux|nul|com[1-9]|lpt[1-9])(\..*)?", part) for part in parts)): + raise ValueError("目标目录必须是根目录内的相对路径,如 inbox/project-a") + return value + + @field_validator("expires_at") + @classmethod + def validate_expiry(cls, value): + # 无时区的后台输入明确解释为北京时间,接口返回保留时区。 + if value.tzinfo is None: + value = value.replace(tzinfo=timezone(timedelta(hours=8))) + if value <= datetime.now(timezone.utc): + raise ValueError("有效期必须晚于当前时间") + return value + + +class VerifyDeliveryCode(BaseModel): + model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) + code: str = Field(min_length=8, max_length=64) + + +class SetDeliveryEnabled(BaseModel): + model_config = ConfigDict(extra="forbid") + enabled: bool + + +class DeliveryShareOptions(BaseModel): + """上传选择的过期策略仍受全站白名单和最长保存时间约束。""" + expire_style: str = Field(min_length=1, max_length=20) + expire_value: int = Field(default=1, ge=1, le=1000000) diff --git a/apps/delivery/services.py b/apps/delivery/services.py new file mode 100644 index 000000000..a5af649f9 --- /dev/null +++ b/apps/delivery/services.py @@ -0,0 +1,333 @@ +"""寄件业务:独立权限、数据库次数预占、存储落盘和失败回收。""" + +import asyncio +import hashlib +import hmac +import os +import secrets +import uuid +from datetime import timedelta + +from fastapi import HTTPException +from tortoise.exceptions import IntegrityError +from tortoise.expressions import F +from tortoise.transactions import in_transaction + +from apps.admin.dependencies import create_token, verify_token +from apps.base.file_validation import validate_upload_file +from apps.base.models import DeliveryCode, DeliveryFile, FileCodes, KeyValue, UploadChunk, PresignUploadSession, StorageReservation +from apps.base.utils import get_expire_info, validate_expire_style +from apps.base.quota import _sql_placeholders, reserve_storage +from apps.delivery.storage import get_storage, validate_storage_config +from core.logger import logger +from core.settings import settings +from core.storage import StoredFile +from core.utils import get_now, sanitize_filename + +TOKEN_TTL = 900 +STALE_SECONDS = 7200 + + +def code_digest(code: str) -> str: + """校验使用带服务端密钥的摘要,与管理员读取原文的用途分开。""" + secret = str(settings.jwt_secret) + if not secret: + raise HTTPException(503, "系统签名密钥尚未初始化") + return hmac.new(secret.encode(), ("delivery-code:" + code).encode(), hashlib.sha256).hexdigest() + + +def upload_identity(authorization: str | None) -> int: + """只接受用途为 delivery 的凭证;管理员 token 也不能被误当作寄件授权。""" + if not authorization or not authorization.startswith("Bearer "): + raise HTTPException(401, "请先验证寄件码") + try: + payload = verify_token(authorization[7:]) + if payload.get("purpose") != "delivery" or payload.get("is_admin"): + raise ValueError("凭证用途错误") + return int(payload["delivery_id"]) + except (ValueError, TypeError, KeyError): + raise HTTPException(401, "寄件凭证无效或已过期,请重新验证寄件码") from None + + +async def active_code(code_id: int) -> DeliveryCode: + record = await DeliveryCode.filter(id=code_id, owner_id="admin", enabled=True, deleted=False).first() + if not record or record.expires_at <= await get_now(): + raise HTTPException(403, "寄件码无效、已过期或已停用") + return record + + +async def create_code(data): + """创建时保存原文,便于管理员后续查看;访客响应仍不提供任何口令列表。""" + validate_storage_config(data.storage_type) + code = data.code or "".join(secrets.choice("ABCDEFGHJKLMNPQRSTUVWXYZ23456789") for _ in range(16)) + try: + record = await DeliveryCode.create( + code_digest=code_digest(code), code_value=code, name=data.name, owner_id="admin", + storage_type=data.storage_type, target_path=data.target_path, + expires_at=data.expires_at, max_uploads=data.max_uploads, + ) + except IntegrityError: + raise HTTPException(409, "该寄件码已被使用,请设置其他口令") from None + return {"item": await code_summary(record), "code": code} + + +async def code_summary(record): + """仅供已鉴权的后台读取状态和口令原文,不返回摘要或存储密钥。""" + now = await get_now() + state = "active" + if record.deleted: + state = "deleted" + elif not record.enabled: + state = "disabled" + elif record.expires_at <= now: + state = "expired" + elif record.used_count >= record.max_uploads: + state = "exhausted" + return { + "id": record.id, "name": record.name, "storage_type": record.storage_type, + "code": record.code_value, + "target_path": record.target_path, "expires_at": record.expires_at, + "max_uploads": record.max_uploads, "used_count": record.used_count, + "reserved_count": record.reserved_count, "enabled": record.enabled, + "deleted": record.deleted, "status": state, "created_at": record.created_at, + "remaining": max(0, record.max_uploads - record.used_count - record.reserved_count), + } + + +async def verify_code(code: str): + record = await DeliveryCode.filter(code_digest=code_digest(code)).first() + if not record: + raise HTTPException(403, "寄件码无效、已过期或已停用") + record = await active_code(record.id) + remaining = record.max_uploads - record.used_count - record.reserved_count + # 已预占的分片会话允许重新验证后续传,新文件仍由 reserve_slot 拒绝超额。 + if remaining <= 0 and not await DeliveryFile.filter(delivery_id=record.id, status="pending", token__startswith="d_").exists(): + raise HTTPException(409, "可上传次数已耗尽或正在使用,请联系管理员") + # 旧码无法离线还原;持有者成功验证时补存其原码,不修改口令或重新生成。 + if record.code_value is None: + await DeliveryCode.filter(id=record.id, code_value__isnull=True).update(code_value=code) + # 凭证仅含寄件 ID,不携带 is_admin、目标路径或下载口令。 + token = create_token({"purpose": "delivery", "delivery_id": record.id}, expires_in=TOKEN_TTL) + return { + "token": token, "expires_in": TOKEN_TTL, "name": record.name, + "remaining": remaining, "expires_at": record.expires_at, + "upload_size": settings.upload_size, "allowed_file_types": settings.allowed_file_types, + "expire_style": settings.expire_style, "max_save_seconds": settings.max_save_seconds, + "enable_chunk": settings.enable_chunk, + } + + +async def reserve_slot(code_id: int): + """数据库原子预占最后一次上传;不以进程内锁代替跨 worker 并发控制。""" + async with in_transaction() as conn: + now = await get_now() + p = _sql_placeholders(2) + count, _ = await conn.execute_query( + f"UPDATE deliverycode SET reserved_count = reserved_count + 1 " + f"WHERE id = {p[0]} AND expires_at > {p[1]} AND owner_id = 'admin' " + "AND enabled = 1 AND deleted = 0 AND used_count + reserved_count < max_uploads", + [code_id, now], + ) + if count != 1: + raise HTTPException(409, "寄件码已失效或没有剩余上传次数") + code = await DeliveryCode.get(id=code_id).using_db(conn) + return await DeliveryFile.create( + delivery_id=code.id, owner_id="admin", token=uuid.uuid4().hex, + file_path=code.target_path, storage_type=code.storage_type, using_db=conn, + ) + + +def stored_file(record): + """显示名保留,磁盘/对象键使用唯一名称,禁止同名覆盖。""" + prefix, suffix = os.path.splitext(record.filename) + return StoredFile(file_path=record.file_path, uuid_file_name=record.stored_name, prefix=prefix, suffix=suffix) + + +async def store_upload(record, file, *, expire_style=None, expire_value=1): + """继承类型与容量限制;只在落盘和记录事务都成功后扣减成功次数。""" + # 传入过期策略即明确请求生成取件码;旧客户端未传策略时仍保持私有收件语义。 + if expire_style is not None: + validate_expire_style(expire_style) + await get_expire_info(expire_value, expire_style) + await validate_upload_file(file) + size = file.size + if size is None: + file.file.seek(0, 2) + size = file.file.tell() + await file.seek(0) + if size > int(settings.upload_size): + raise HTTPException(413, "文件大小超过站点限制") + filename = await normalize_delivery_filename(file.filename) + record.filename = filename + record.stored_name = f"{record.token}_{filename}" + record.size = size + await reserve_storage("delivery:" + record.token, size, STALE_SECONDS) + await record.save(update_fields=["filename", "stored_name", "size", "updated_at"]) + storage = await get_storage(record.storage_type) + # shield 防止本地后台写线程被取消后仍写入已经关闭的临时文件。 + writing = asyncio.create_task(storage.save_file(file.file, f"{record.file_path}/{record.stored_name}", file.content_type)) + try: + await asyncio.shield(writing) + except asyncio.CancelledError: + try: + await writing + except Exception: + logger.warning("被取消的寄件写入失败 id=%s", record.id, exc_info=True) + raise + # 兼容接口仅负责构造分享字段,扣次和关联提交与普通上传共用同一事务。 + share_fields = None + if expire_style is not None: + expired_at, expired_count, used_count, code = await get_expire_info(expire_value, expire_style) + share_fields = dict(code=code, size=size, expired_at=expired_at, + expired_count=expired_count, used_count=used_count) + share = await commit_delivery(record, share_fields) + result = {"name": filename, "size": size, "message": "投递成功"} + if share is not None: + result.update(code=share.code, expired_at=share.expired_at, expired_count=share.expired_count) + return result + + +async def heartbeat(record): + """有效传输定期续租,进程异常退出后遗留记录才会被回收。""" + while True: + await asyncio.sleep(30) + changed = await DeliveryFile.filter(id=record.id, status__in=["pending", "finalizing"]).update(updated_at=await get_now()) + if not changed: + return + # 大文件慢速传输期间维持容量预留,避免 TTL 到期使其他上传超配额。 + await StorageReservation.filter(token__in=reservation_tokens(record.token)).update( + expires_at=await get_now() + timedelta(seconds=STALE_SECONDS) + ) + + +async def abort_upload(record_id: int, *, stale_before=None): + """先释放次数,再保留 cleanup 记录跟踪残留文件,清理成功后释放占用容量。""" + async with in_transaction() as conn: + query = DeliveryFile.filter(id=record_id, status__in=["pending", "finalizing"]) + if stale_before is not None: + query = query.filter(updated_at__lt=stale_before) + changed = await query.using_db(conn).update(status="cleanup", updated_at=await get_now()) + if changed: + record = await DeliveryFile.get(id=record_id).using_db(conn) + await DeliveryCode.filter(id=record.delivery_id, reserved_count__gt=0).using_db(conn).update( + reserved_count=F("reserved_count") - 1 + ) + # 转为 cleanup 后由收件记录计费,统一释放所有上传预留。 + await StorageReservation.filter(token__in=reservation_tokens(record.token)).using_db(conn).delete() + await clean_file(record_id) + + +async def clean_file(record_id): + """实际文件删除成功后物理删除收件记录;失败时保留记录计费并重试。""" + record = await DeliveryFile.filter(id=record_id, status="cleanup").first() + if not record: + return + try: + if record.stored_name: + storage = await get_storage(record.storage_type) + # 复用普通上传的会话时,临时分片也必须按寄件存储回收,失败保留记录重试。 + if record.token.startswith("d_"): + await storage.clean_chunks(record.token, f"{record.file_path}/{record.stored_name}") + await storage.delete_file(stored_file(record)) + if record.token.startswith("d_"): + await UploadChunk.filter(upload_id=record.token).delete() + await PresignUploadSession.filter(upload_id=record.token).delete() + await StorageReservation.filter(token__in=reservation_tokens(record.token)).delete() + # 与普通文件管理一致,清理成功后不保留已删除文件的历史空壳。 + await DeliveryFile.filter(id=record.id, status="cleanup").delete() + except Exception: + logger.warning("寄件文件清理失败,将自动重试 id=%s", record.id, exc_info=True) + + +async def cleanup_once(): + """有限批次回收崩溃残留,避免任务持有大量 ORM 对象。""" + before = await get_now() - timedelta(seconds=STALE_SECONDS) + for record in await DeliveryFile.filter(status__in=["pending", "finalizing"], updated_at__lt=before).limit(100): + await abort_upload(record.id, stale_before=before) + for record in await DeliveryFile.filter(status="cleanup").limit(100): + await clean_file(record.id) + # 分享创建成功后若分片回收失败,继续清理临时分片,不撤销已经生成的取件码。 + tokens = await UploadChunk.filter(chunk_index=-1, upload_id__startswith="d_").limit(100).values_list("upload_id", flat=True) + for record in await DeliveryFile.filter(token__in=tokens, status="shared"): + try: + storage = await get_storage(record.storage_type) + await storage.clean_chunks(record.token, f"{record.file_path}/{record.stored_name}") + await UploadChunk.filter(upload_id=record.token).delete() + except Exception: + logger.warning("寄件分享临时分片清理失败 id=%s", record.id, exc_info=True) + # 旧版本仅在物理清理成功后标记 deleted 并将大小归零,分批移除这些历史空壳。 + deleted_ids = await DeliveryFile.filter(status="deleted", size=0).limit(100).values_list("id", flat=True) + if deleted_ids: + await DeliveryFile.filter(id__in=deleted_ids, status="deleted", size=0).delete() + + +async def cleanup_loop(): + while True: + try: + await cleanup_once() + except Exception: + logger.warning("寄件清理任务异常,下轮继续", exc_info=True) + await asyncio.sleep(60) + + +async def request_file_removal(record_id): + """普通文件管理、寄件管理和过期清理共用撤销流程,容量始终只计算一次。""" + async with in_transaction() as conn: + record = await DeliveryFile.filter(id=record_id).using_db(conn).first() + # 管理员删除与自动过期可能同时触发;已被另一流程清除时视为完成。 + if record is None: + return + if record.status in {"pending", "finalizing"}: + raise HTTPException(409, "文件正在上传,请稍后再试") + if record.share_id is not None: + await FileCodes.filter(id=record.share_id).using_db(conn).delete() + # 元数据清理由原文件服务管理;这里仅清除对应记录,不能触碰其他分享。 + await KeyValue.filter(key=f"admin_file_metadata:{record.share_id}").using_db(conn).delete() + await DeliveryFile.filter(id=record_id, status__in=["stored", "shared"]).using_db(conn).update( + status="cleanup", updated_at=await get_now() + ) + await clean_file(record_id) + + + +def reservation_tokens(token): + """集中定义寄件会话的容量键,续租、提交和取消不会漏掉某一种上传方式。""" + tokens = ["delivery:" + token] + if token.startswith("d_"): + tokens.extend(["chunk:" + token, "presign:" + token]) + return tokens + + +async def commit_delivery(record, share_fields=None): + """统一提交寄件扣次;无分享字段时保留旧客户端的私有收件语义。""" + async with in_transaction() as conn: + changed = await DeliveryFile.filter(id=record.id, status__in=["pending", "finalizing"]).using_db(conn).update( + status="shared" if share_fields is not None else "stored", updated_at=await get_now() + ) + if changed != 1: + raise HTTPException(409, "该上传已完成或已被清理") + changed = await DeliveryCode.filter( + id=record.delivery_id, enabled=True, deleted=False, + expires_at__gt=await get_now(), reserved_count__gt=0, + ).using_db(conn).update(reserved_count=F("reserved_count") - 1, used_count=F("used_count") + 1) + if changed != 1: + raise HTTPException(409, "寄件码在上传期间失效") + share = None + if share_fields is not None: + fields = dict(share_fields) + # 存储名保证唯一,显示名、文本分享和原取件规则保持不变。 + if "text" not in fields: + fields["prefix"], fields["suffix"] = os.path.splitext(record.filename) + fields["file_path"] = record.file_path + fields["uuid_file_name"] = record.stored_name + share = await FileCodes.create(using_db=conn, **fields) + await DeliveryFile.filter(id=record.id).using_db(conn).update(share_id=share.id, size=share.size) + await StorageReservation.filter(token__in=reservation_tokens(record.token)).using_db(conn).delete() + return share + + + +async def normalize_delivery_filename(file_name): + """新旧上传统一清理显示名,并预留唯一前缀所需的文件系统字节空间。""" + filename = await sanitize_filename((file_name or "file").replace("\\", "/").split("/")[-1]) + return filename.encode("utf-8")[:180].decode("utf-8", errors="ignore") or "file" diff --git a/apps/delivery/static/admin.html b/apps/delivery/static/admin.html new file mode 100644 index 000000000..625b3d718 --- /dev/null +++ b/apps/delivery/static/admin.html @@ -0,0 +1,48 @@ + + +
+ +
+ + + + +
+