严格隔离: 每账户独立空间, admin 不再全览 (只 admin/无主旧文件); 测试更新

This commit is contained in:
lou
2026-08-10 18:45:22 +08:00
parent 4be951fe4a
commit a81fcfdefa
3 changed files with 18 additions and 7 deletions
+3 -1
View File
@@ -60,8 +60,10 @@ def verify_token(authorization: str | None = Header(default=None)) -> str:
def _check_owner(rec: dict, username: str) -> None:
"""文件归属校验: admin 可操作全部, 普通用户只能操作自己的"""
"""文件归属校验: 每个账户只操作自己的空间 (admin 只认 admin/无主旧文件)"""
if username == "admin":
if rec.get("user_id") not in ("admin", None):
raise HTTPException(status_code=403, detail="无权访问他人文件")
return
if rec.get("user_id") != username:
raise HTTPException(status_code=403, detail="无权访问他人文件")
+8 -4
View File
@@ -258,12 +258,13 @@ class ServerDB:
return dict(row) if row else None
def list_files(self, user_id: str) -> list[dict[str, Any]]:
"""文件记录 (ls 端点): admin 看全部, 普通用户看自己的"""
"""文件记录 (ls 端点): 每个账户只看自己的空间 (admin 含无主旧文件)"""
with self._conn() as conn:
if user_id == "admin":
rows = conn.execute(
"SELECT file_id, file_name, size, sha256, created_at FROM files "
"ORDER BY created_at DESC"
"WHERE user_id = ? OR user_id IS NULL ORDER BY created_at DESC",
("admin",),
).fetchall()
else:
rows = conn.execute(
@@ -280,10 +281,13 @@ class ServerDB:
return cur.rowcount > 0
def sum_files_size(self, user_id: str) -> int:
"""文件 size 总和 (配额统计用): admin 全部, 普通用户自己的"""
"""文件 size 总和 (配额统计用): 每个账户只算自己的空间"""
with self._conn() as conn:
if user_id == "admin":
row = conn.execute("SELECT COALESCE(SUM(size), 0) FROM files").fetchone()
row = conn.execute(
"SELECT COALESCE(SUM(size), 0) FROM files WHERE user_id = ? OR user_id IS NULL",
("admin",),
).fetchone()
else:
row = conn.execute(
"SELECT COALESCE(SUM(size), 0) FROM files WHERE user_id = ?",
+7 -2
View File
@@ -388,6 +388,11 @@ class AccountTest(unittest.TestCase):
# alice 自己可以下载/删除
r = self.client.get(f"/api/files/{file_id}/chunk/1", headers=auth_a)
self.assertEqual(r.status_code, 200)
# admin (SZ_TOKEN) 可见全部
# admin (SZ_TOKEN) 也只看自己的空间, 看不到 alice 的文件
r = self.client.get("/api/files", headers=AUTH)
self.assertTrue(any(f["file_id"] == file_id for f in r.json()["files"]))
self.assertTrue(all(f["file_id"] != file_id for f in r.json()["files"]))
# admin 下载/删除 alice 的文件 -> 403
r = self.client.get(f"/api/files/{file_id}/chunk/1", headers=AUTH)
self.assertEqual(r.status_code, 403)
r = self.client.delete(f"/api/files/{file_id}", headers=AUTH)
self.assertEqual(r.status_code, 403)