feat: backup functionality (untested)

Author: tokenicrat

src/pelit/plib/route_tool.py

@@ -4,9 +4,7 @@
 from typing import Optional, Any
 from pathlib import Path
 from flask import request
-
-__all__ = ['authenticate', 'generate_file_name', 'enough_space',
-           'join_url', 'list_dir', 'is_attempting_traversal']
+import shutil
 
 def authenticate(cfg: dict[str, Any]) -> bool:
     """
@@ -102,3 +100,17 @@ def is_attempting_traversal(comp: str) -> bool:
     if '..' in comp or '/' in comp or '\\' in comp:
         return True
     return False
+
+def backup_to_file(path: Path, file: Path) -> None:
+    """
+    创建全目录备份
+
+    Args:
+        path: 需要备份的目录
+        file: 备份保存的文件
+    """
+
+    try:
+        shutil.make_archive(str(path), 'gztar', str(file))
+    except Exception:
+        pass

src/pelit/route.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 from pelit.plib.log import p_logger
 from pelit.plib.route_tool import *
+from multiprocessing import Process
 
 def create_route(cfg: dict[str, Any], lg: p_logger) -> Blueprint:
     """
@@ -172,45 +173,14 @@ def _delete(directory: str, file: str) -> tuple[Response, int]:
             }), 502
 
     @main_route.route('/list', methods=['GET'])
-    def _list_root() -> tuple[Response, int]:
-        """
-        列出根目录下所有目录
-
-        Returns:
-            JSON 格式的列表和响应码
-        """
-        info_head = f"{request.remote_addr} {request.method} {request.path}"
-
-        if not authenticate(cfg):
-            lg.warn(f"{info_head} 401: 认证失败")
-            return jsonify({
-                "success": False,
-                "message": "认证失败"
-            }), 401
-        
-        try:
-            storage_path = Path(cfg['storage']['path'])
-            resp: dict[str, Any] = {
-                "success": True,
-                "message": "",
-                "list": list_dir(storage_path)
-            }
-            lg.info(f'{info_head} 200 列举成功')
-            return jsonify(resp), 200
-        except Exception as e:
-            lg.warn(f'{info_head} 502 列举失败')
-            lg.warn('这是一个内部错误，请检查配置')
-            lg.warn(f'{e}')
-            return jsonify({
-                "success": False,
-                "message": "列举失败"
-            }), 502
-        
     @main_route.route('/list/<directory>', methods=['GET'])
-    def _list_dir(directory: str) -> tuple[Response, int]:
+    def _list(directory: str = "/") -> tuple[Response, int]:
         """
         列出指定目录下所有目录
 
+        Args:
+            directory: 列举地址，未指定时默认为根目录
+
         Returns:
             JSON 格式的列表和响应码
         """
@@ -223,6 +193,10 @@ def _list_dir(directory: str) -> tuple[Response, int]:
                 "message": "认证失败"
             }), 401
 
+        if is_attempting_traversal(directory):
+            lg.warn(f"{info_head} 403 危险请求")
+            return Response("禁止访问"), 403
+
         try:
             path = Path(cfg['storage']['path']) / directory
             resp: dict[str, Any] = {
@@ -306,4 +280,33 @@ def _retrieve(directory: str, file: str) -> tuple[Response, int]:
             lg.warn(f"{e}")
             return Response("内部错误"), 502
 
+    @main_route.route('/backup', methods=['GET'])
+    @main_route.route('/backup/<directory>', methods=['GET'])
+    def _backup(directory: str = '/') -> tuple[Response, int]:
+        """
+        创建一个备份任务，具体方式在配置文件中指定
+
+        Returns:
+            包含响应和状态码
+        """
+        info_head = f"{request.remote_addr} {request.method} {request.path}"
+
+        bak_name = generate_file_name(cfg['storage']['path'], '.tar.gz')
+        bak_path = Path(cfg['storage']['path']) / (bak_name + '.tar.gz')
+        bak_url = join_url(cfg['network']['base_url'], bak_name + '.tar.gz')
+
+        path = Path(cfg['storage']['path']) / directory
+
+        # 创建新进程压缩备份
+        # FIXME: 由于是多进程的，这个任务失败了也不会有表示
+        p = Process(target=backup_to_file, args=(path, bak_path))
+        p.start()
+
+        lg.info(f"{info_head} 200 备份任务创建成功")
+        return jsonify({
+            "success": True,
+            "url": bak_url,
+            "message": "备份任务创建成功"
+        }), 200
+
     return main_route