feat: 支持文件上传下载备份 (#10438)

* feat: 支持文件上传下载备份

* perf: 抽离replay和ftpfile存储代码

* perf: FTPLog增加session字段

* fix: 修改变量名

apps/audits/api.py

@@ -1,30 +1,47 @@
 # -*- coding: utf-8 -*-
 #
+import os
+
 from importlib import import_module
 
 from django.conf import settings
+from django.shortcuts import get_object_or_404
 from django.db.models import F, Value, CharField, Q
+from django.http import HttpResponse, FileResponse
+from django.utils.encoding import escape_uri_path
 from rest_framework import generics
 from rest_framework.permissions import IsAuthenticated
+from rest_framework.response import Response
+from rest_framework.views import APIView
+from rest_framework.decorators import action
 
+from common.api import AsyncApiMixin
 from common.drf.filters import DatetimeRangeFilter
+from common.permissions import IsServiceAccount
 from common.plugins.es import QuerySet as ESQuerySet
-from common.utils import is_uuid
-from common.utils import lazyproperty
+from common.utils import is_uuid, get_logger, lazyproperty
+from common.const.http import GET, POST
+from common.storage.ftp_file import FTPFileStorageHandler
 from orgs.mixins.api import OrgReadonlyModelViewSet, OrgModelViewSet
 from orgs.utils import current_org, tmp_to_root_org
 from orgs.models import Organization
+from rbac.permissions import RBACPermission
+from terminal.models import default_storage
 from users.models import User
 from .backends import TYPE_ENGINE_MAPPING
 from .const import ActivityChoices
 from .models import FTPLog, UserLoginLog, OperateLog, PasswordChangeLog, ActivityLog, JobLog
-from .serializers import FTPLogSerializer, UserLoginLogSerializer, JobLogSerializer
 from .serializers import (
+    FTPLogSerializer, UserLoginLogSerializer, JobLogSerializer,
     OperateLogSerializer, OperateLogActionDetailSerializer,
     PasswordChangeLogSerializer, ActivityUnionLogSerializer,
+    FileSerializer
 )
 
 
+logger = get_logger(__name__)
+
+
 class JobAuditViewSet(OrgReadonlyModelViewSet):
     model = JobLog
     extra_filter_backends = [DatetimeRangeFilter]
@@ -47,7 +64,49 @@ class FTPLogViewSet(OrgModelViewSet):
     filterset_fields = ['user', 'asset', 'account', 'filename']
     search_fields = filterset_fields
     ordering = ['-date_start']
-    http_method_names = ['post', 'get', 'head', 'options']
+    http_method_names = ['post', 'get', 'head', 'options', 'patch']
+    rbac_perms = {
+        'download': 'audits.view_ftplog',
+    }
+
+    def get_storage(self):
+        return FTPFileStorageHandler(self.get_object())
+
+    @action(
+        methods=[GET], detail=True, permission_classes=[RBACPermission, ],
+        url_path='file/download'
+    )
+    def download(self, request, *args, **kwargs):
+        ftp_log = self.get_object()
+        ftp_storage = self.get_storage()
+        local_path, url_or_err = ftp_storage.get_file_path_url()
+        if local_path is None:
+            return HttpResponse(url_or_err)
+
+        file = open(default_storage.path(local_path), 'rb')
+        response = FileResponse(file)
+        response['Content-Type'] = 'application/octet-stream'
+        filename = escape_uri_path(ftp_log.filename)
+        response["Content-Disposition"] = "attachment; filename*=UTF-8''{}".format(filename)
+        return response
+
+    @action(methods=[POST], detail=True, permission_classes=[IsServiceAccount, ], serializer_class=FileSerializer)
+    def upload(self, request, *args, **kwargs):
+        ftp_log = self.get_object()
+        serializer = self.get_serializer(data=request.data)
+        if serializer.is_valid():
+            file = serializer.validated_data['file']
+            name, err = ftp_log.save_file_to_storage(file)
+            if not name:
+                msg = "Failed save file `{}`: {}".format(ftp_log.id, err)
+                logger.error(msg)
+                return Response({'msg': str(err)}, status=400)
+            url = default_storage.url(name)
+            return Response({'url': url}, status=201)
+        else:
+            msg = 'Upload data invalid: {}'.format(serializer.errors)
+            logger.error(msg)
+            return Response({'msg': serializer.errors}, status=401)
 
 
 class UserLoginCommonMixin:

apps/audits/const.py

@@ -16,6 +16,7 @@ class OperateChoices(TextChoices):
     rename = "rename", _("Rename")
     symlink = "symlink", _("Symlink")
     download = "download", _("Download")
+    rename_dir = "rename_dir", _("Rename dir")
 
 
 class ActionChoices(TextChoices):

apps/audits/migrations/0022_auto_20230605_1555.py

@@ -0,0 +1,29 @@
+# Generated by Django 3.2.17 on 2023-06-05 07:55
+
+from django.db import migrations, models
+import uuid
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('audits', '0021_auto_20230207_0857'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='ftplog',
+            name='has_file',
+            field=models.BooleanField(default=False, verbose_name='File Record'),
+        ),
+        migrations.AddField(
+            model_name='ftplog',
+            name='session',
+            field=models.CharField(default=uuid.uuid4, max_length=36, verbose_name='Session'),
+        ),
+        migrations.AlterField(
+            model_name='ftplog',
+            name='operate',
+            field=models.CharField(choices=[('mkdir', 'Mkdir'), ('rmdir', 'Rmdir'), ('delete', 'Delete'), ('upload', 'Upload'), ('rename', 'Rename'), ('symlink', 'Symlink'), ('download', 'Download'), ('rename_dir', 'Rename dir')], max_length=16, verbose_name='Operate'),
+        ),
+    ]

apps/audits/models.py

@@ -1,7 +1,9 @@
+import os
 import uuid
 
 from django.db import models
 from django.db.models import Q
+from django.conf import settings
 from django.utils import timezone
 from django.utils.translation import gettext, ugettext_lazy as _
 
@@ -10,6 +12,7 @@ from common.utils import lazyproperty
 from ops.models import JobExecution
 from orgs.mixins.models import OrgModelMixin, Organization
 from orgs.utils import current_org
+from terminal.models import default_storage
 from .const import (
     OperateChoices,
     ActionChoices,
@@ -40,6 +43,8 @@ class JobLog(JobExecution):
 
 
 class FTPLog(OrgModelMixin):
+    upload_to = 'FTP_FILES'
+
     id = models.UUIDField(default=uuid.uuid4, primary_key=True)
     user = models.CharField(max_length=128, verbose_name=_("User"))
     remote_addr = models.CharField(
@@ -53,10 +58,27 @@ class FTPLog(OrgModelMixin):
     filename = models.CharField(max_length=1024, verbose_name=_("Filename"))
     is_success = models.BooleanField(default=True, verbose_name=_("Success"))
     date_start = models.DateTimeField(auto_now_add=True, verbose_name=_("Date start"))
+    has_file = models.BooleanField(default=False, verbose_name=_("File Record"))
+    session = models.CharField(max_length=36, verbose_name=_("Session"), default=uuid.uuid4)
 
     class Meta:
         verbose_name = _("File transfer log")
 
+    @property
+    def filepath(self):
+        return os.path.join(self.upload_to, self.date_start.strftime('%Y-%m-%d'), str(self.id))
+
+    def save_file_to_storage(self, file):
+        try:
+            name = default_storage.save(self.filepath, file)
+        except OSError as e:
+            return None, e
+
+        if settings.SERVER_REPLAY_STORAGE:
+            from .tasks import upload_ftp_file_to_external_storage
+            upload_ftp_file_to_external_storage.delay(str(self.id), file.name)
+        return name, None
+
 
 class OperateLog(OrgModelMixin):
     id = models.UUIDField(default=uuid.uuid4, primary_key=True)

apps/audits/serializers.py

@@ -37,8 +37,8 @@ class FTPLogSerializer(serializers.ModelSerializer):
         fields_mini = ["id"]
         fields_small = fields_mini + [
             "user", "remote_addr", "asset", "account",
-            "org_id", "operate", "filename", "is_success",
-            "date_start",
+            "org_id", "operate", "filename", "date_start",
+            "is_success", "has_file",
         ]
         fields = fields_small
 
@@ -158,3 +158,7 @@ class ActivityUnionLogSerializer(serializers.Serializer):
                 api_to_ui=True, is_audit=True
             )
         return detail_url
+
+
+class FileSerializer(serializers.Serializer):
+    file = serializers.FileField(allow_empty_file=True)

apps/audits/tasks.py

@@ -11,13 +11,16 @@ from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 
 from common.utils import get_log_keep_day, get_logger
+from common.storage.ftp_file import FTPFileStorageHandler
 from ops.celery.decorator import (
     register_as_period_task, after_app_shutdown_clean_periodic
 )
 from ops.models import CeleryTaskExecution
 from terminal.models import Session, Command
+from terminal.backends import server_replay_storage
 from .models import UserLoginLog, OperateLog, FTPLog, ActivityLog
 
+
 logger = get_logger(__name__)
 
 
@@ -46,7 +49,15 @@ def clean_ftp_log_period():
     now = timezone.now()
     days = get_log_keep_day('FTP_LOG_KEEP_DAYS')
     expired_day = now - datetime.timedelta(days=days)
+    file_store_dir = os.path.join(default_storage.base_location, 'ftp_file')
     FTPLog.objects.filter(date_start__lt=expired_day).delete()
+    command = "find %s -mtime +%s -exec rm -f {} \\;" % (
+        file_store_dir, days
+    )
+    subprocess.call(command, shell=True)
+    command = "find %s -type d -empty -delete;" % file_store_dir
+    subprocess.call(command, shell=True)
+    logger.info("Clean FTP file done")
 
 
 def clean_celery_tasks_period():
@@ -98,3 +109,27 @@ def clean_audits_log_period():
     clean_activity_log_period()
     clean_celery_tasks_period()
     clean_expired_session_period()
+
+
+@shared_task(verbose_name=_('Upload FTP file to external storage'))
+def upload_ftp_file_to_external_storage(ftp_log_id, file_name):
+    logger.info(f'Start upload FTP file record to external storage: {ftp_log_id} - {file_name}')
+    ftp_log = FTPLog.objects.filter(id=ftp_log_id).first()
+    if not ftp_log:
+        logger.error(f'FTP db item not found: {ftp_log_id}')
+        return
+    ftp_storage = FTPFileStorageHandler(ftp_log)
+    local_path, url_or_err = ftp_storage.find_local()
+    if not local_path:
+        logger.error(f'FTP file record not found, may be upload error. file name: {file_name}')
+        return
+    abs_path = default_storage.path(local_path)
+    ok, err = server_replay_storage.upload(abs_path, ftp_log.filepath)
+    if not ok:
+        logger.error(f'Session file record upload to external error: {err}')
+        return
+    try:
+        default_storage.delete(local_path)
+    except:
+        pass
+    return

apps/audits/utils.py

@@ -1,7 +1,5 @@
 import codecs
 import copy
-import csv
-
 from itertools import chain
 from datetime import datetime
 

apps/common/storage/base.py

@@ -0,0 +1,65 @@
+import os
+
+import jms_storage
+
+from django.conf import settings
+
+from terminal.models import default_storage, ReplayStorage
+from common.utils import get_logger, make_dirs
+
+
+logger = get_logger(__name__)
+
+
+class BaseStorageHandler(object):
+    NAME = ''
+
+    def __init__(self, obj):
+        self.obj = obj
+
+    def get_file_path(self, **kwargs):
+        # return remote_path, local_path
+        raise NotImplementedError
+
+    def find_local(self):
+        raise NotImplementedError
+
+    def download(self):
+        replay_storages = ReplayStorage.objects.all()
+        configs = {
+            storage.name: storage.config
+            for storage in replay_storages
+            if not storage.type_null_or_server
+        }
+        if settings.SERVER_REPLAY_STORAGE:
+            configs['SERVER_REPLAY_STORAGE'] = settings.SERVER_REPLAY_STORAGE
+        if not configs:
+            msg = f"Not found {self.NAME} file, and not remote storage set"
+            return None, msg
+        storage = jms_storage.get_multi_object_storage(configs)
+
+        remote_path, local_path = self.get_file_path(storage=storage)
+        if not remote_path:
+            msg = f'Not found {self.NAME} file'
+            logger.error(msg)
+            return None, msg
+
+        # 保存到storage的路径
+        target_path = os.path.join(default_storage.base_location, local_path)
+        target_dir = os.path.dirname(target_path)
+        if not os.path.isdir(target_dir):
+            make_dirs(target_dir, exist_ok=True)
+
+        ok, err = storage.download(remote_path, target_path)
+        if not ok:
+            msg = f'Failed download {self.NAME} file: {err}'
+            logger.error(msg)
+            return None, msg
+        url = default_storage.url(local_path)
+        return local_path, url
+
+    def get_file_path_url(self):
+        local_path, url = self.find_local()
+        if local_path is None:
+            local_path, url = self.download()
+        return local_path, url

apps/common/storage/ftp_file.py

@@ -0,0 +1,17 @@
+from terminal.models import default_storage
+from .base import BaseStorageHandler
+
+
+class FTPFileStorageHandler(BaseStorageHandler):
+    NAME = 'FTP'
+
+    def get_file_path(self, **kwargs):
+        return self.obj.filepath, self.obj.filepath
+
+    def find_local(self):
+        local_path = self.obj.filepath
+        # 去default storage中查找
+        if default_storage.exists(local_path):
+            url = default_storage.url(local_path)
+            return local_path, url
+        return None, None

apps/common/storage/replay.py

@@ -0,0 +1,31 @@
+from itertools import chain
+
+from terminal.models import default_storage
+from .base import BaseStorageHandler
+
+
+class ReplayStorageHandler(BaseStorageHandler):
+    NAME = 'REPLAY'
+
+    def get_file_path(self, **kwargs):
+        storage = kwargs['storage']
+        # 获取外部存储路径名
+        session_path = self.obj.find_ok_relative_path_in_storage(storage)
+        if not session_path:
+            return None
+
+        # 通过外部存储路径名后缀，构造真实的本地存储路径
+        return session_path, self.obj.get_local_path_by_relative_path(session_path)
+
+    def find_local(self):
+        # 存在外部存储上，所有可能的路径名
+        session_paths = self.obj.get_all_possible_relative_path()
+
+        # 存在本地存储上，所有可能的路径名
+        local_paths = self.obj.get_all_possible_local_path()
+
+        for _local_path in chain(session_paths, local_paths):
+            if default_storage.exists(_local_path):
+                url = default_storage.url(_local_path)
+                return _local_path, url
+        return None, f'{self.NAME} not found.'

apps/jumpserver/conf.py

@@ -548,6 +548,9 @@ class Config(dict):
 
         # Applet 等软件的下载地址
         'APPLET_DOWNLOAD_HOST': '',
+
+        # FTP 文件上传下载备份阈值，单位(M)，当值小于等于0时，不备份
+        'FTP_FILE_MAX_STORE': 100,
     }
 
     def __init__(self, *args):

apps/jumpserver/settings/custom.py

@@ -29,6 +29,7 @@ DEFAULT_TERMINAL_REPLAY_STORAGE = {
     },
 }
 TERMINAL_REPLAY_STORAGE = CONFIG.TERMINAL_REPLAY_STORAGE
+FTP_FILE_MAX_STORE = CONFIG.FTP_FILE_MAX_STORE
 
 # Security settings
 SECURITY_MFA_AUTH = CONFIG.SECURITY_MFA_AUTH

apps/rbac/const.py

@@ -91,7 +91,7 @@ exclude_permissions = (
     ('audits', 'activitylog', 'add,delete,change', 'activitylog'),
     ('audits', 'passwordchangelog', 'add,change,delete', 'passwordchangelog'),
     ('audits', 'userloginlog', 'add,change,delete,change', 'userloginlog'),
-    ('audits', 'ftplog', 'change,delete', 'ftplog'),
+    ('audits', 'ftplog', 'delete', 'ftplog'),
     ('tickets', 'ticketassignee', '*', 'ticketassignee'),
     ('tickets', 'ticketflow', 'add,delete', 'ticketflow'),
     ('tickets', 'comment', '*', '*'),

apps/terminal/api/session/session.py

@@ -22,15 +22,13 @@ from common.drf.renders import PassthroughRenderer
 from common.api import AsyncApiMixin
 from common.utils import data_to_json, is_uuid
 from common.utils import get_logger, get_object_or_none
+from common.storage.replay import ReplayStorageHandler
 from rbac.permissions import RBACPermission
 from orgs.mixins.api import OrgBulkModelViewSet
 from orgs.utils import tmp_to_root_org, tmp_to_org
 from terminal import serializers
 from terminal.models import Session
-from terminal.utils import (
-    find_session_replay_local, download_session_replay,
-    is_session_approver, get_session_replay_url
-)
+from terminal.utils import is_session_approver
 from terminal.permissions import IsSessionAssignee
 from users.models import User
 
@@ -112,20 +110,23 @@ class SessionViewSet(OrgBulkModelViewSet):
         os.chdir(current_dir)
         return file
 
+    def get_storage(self):
+        return ReplayStorageHandler(self.get_object())
+
     @action(methods=[GET], detail=True, renderer_classes=(PassthroughRenderer,), url_path='replay/download',
             url_name='replay-download')
     def download(self, request, *args, **kwargs):
-        session = self.get_object()
-        local_path, url = get_session_replay_url(session)
+        storage = self.get_storage()
+        local_path, url_or_err = storage.get_file_path_url()
         if local_path is None:
-            return Response({"error": url}, status=404)
-        file = self.prepare_offline_file(session, local_path)
+            return Response({'error': url_or_err}, status=404)
 
+        file = self.prepare_offline_file(storage.obj, local_path)
         response = FileResponse(file)
         response['Content-Type'] = 'application/octet-stream'
         # 这里要注意哦，网上查到的方法都是response['Content-Disposition']='attachment;filename="filename.py"',
         # 但是如果文件名是英文名没问题，如果文件名包含中文，下载下来的文件名会被改为url中的path。
-        filename = escape_uri_path('{}.tar'.format(session.id))
+        filename = escape_uri_path('{}.tar'.format(storage.obj.id))
         disposition = "attachment; filename*=UTF-8''{}".format(filename)
         response["Content-Disposition"] = disposition
         return response
@@ -208,13 +209,12 @@ class SessionReplayViewSet(AsyncApiMixin, viewsets.ViewSet):
     def retrieve(self, request, *args, **kwargs):
         session_id = kwargs.get('pk')
         session = get_object_or_404(Session, id=session_id)
-        local_path, url = find_session_replay_local(session)
 
-        if not local_path:
-            local_path, url = download_session_replay(session)
-            if not local_path:
-                return Response({"error": url}, status=404)
-        data = self.get_replay_data(session, url)
+        storage = ReplayStorageHandler(session)
+        local_path, url_or_err = storage.get_file_path_url()
+        if url_or_err:
+            return Response({"error": url_or_err}, status=404)
+        data = self.get_replay_data(session, url_or_err)
         return Response(data)
 
 

apps/terminal/models/component/terminal.py

@@ -129,7 +129,8 @@ class Terminal(StorageMixin, TerminalStatusMixin, JMSBaseModel):
         configs.update(self.get_login_title_setting())
         configs.update({
             'SECURITY_MAX_IDLE_TIME': settings.SECURITY_MAX_IDLE_TIME,
-            'SECURITY_SESSION_SHARE': settings.SECURITY_SESSION_SHARE
+            'SECURITY_SESSION_SHARE': settings.SECURITY_SESSION_SHARE,
+            'FTP_FILE_MAX_STORE': settings.FTP_FILE_MAX_STORE,
         })
         return configs
 

apps/terminal/tasks.py

@@ -10,6 +10,7 @@ from django.core.files.storage import default_storage
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 
+from common.storage.replay import ReplayStorageHandler
 from ops.celery.decorator import (
     register_as_period_task, after_app_ready_start,
     after_app_shutdown_clean_periodic
@@ -23,7 +24,6 @@ from .models import (
     AppletHost, ReplayStorage, CommandStorage
 )
 from .notifications import StorageConnectivityMessage
-from .utils import find_session_replay_local
 
 CACHE_REFRESH_INTERVAL = 10
 RUNNING = False
@@ -66,7 +66,8 @@ def upload_session_replay_to_external_storage(session_id):
         logger.error(f'Session db item not found: {session_id}')
         return
 
-    local_path, foobar = find_session_replay_local(session)
+    replay_storage = ReplayStorageHandler(session)
+    local_path, url_or_err = replay_storage.find_local()
     if not local_path:
         logger.error(f'Session replay not found, may be upload error: {local_path}')
         return

apps/terminal/utils/__init__.py

@@ -1,4 +1,3 @@
 from .components import *
 from .common import *
-from .session_replay import *
 from .db_port_mapper import *

apps/terminal/utils/session_replay.py

@@ -1,75 +0,0 @@
-# -*- coding: utf-8 -*-
-#
-import os
-from itertools import groupby, chain
-
-from django.conf import settings
-from django.core.files.storage import default_storage
-
-import jms_storage
-
-from common.utils import get_logger, make_dirs
-from ..models import ReplayStorage
-
-
-logger = get_logger(__name__)
-
-
-def find_session_replay_local(session):
-    # 存在外部存储上，所有可能的路径名
-    session_paths = session.get_all_possible_relative_path()
-
-    # 存在本地存储上，所有可能的路径名
-    local_paths = session.get_all_possible_local_path()
-
-    for _local_path in chain(session_paths, local_paths):
-        if default_storage.exists(_local_path):
-            url = default_storage.url(_local_path)
-            return _local_path, url
-    return None, None
-
-
-def download_session_replay(session):
-    replay_storages = ReplayStorage.objects.all()
-    configs = {
-        storage.name: storage.config
-        for storage in replay_storages
-        if not storage.type_null_or_server
-    }
-    if settings.SERVER_REPLAY_STORAGE:
-        configs['SERVER_REPLAY_STORAGE'] = settings.SERVER_REPLAY_STORAGE
-    if not configs:
-        msg = "Not found replay file, and not remote storage set"
-        return None, msg
-    storage = jms_storage.get_multi_object_storage(configs)
-
-    # 获取外部存储路径名
-    session_path = session.find_ok_relative_path_in_storage(storage)
-    if not session_path:
-        msg = "Not found session replay file"
-        return None, msg
-
-    # 通过外部存储路径名后缀，构造真实的本地存储路径
-    local_path = session.get_local_path_by_relative_path(session_path)
-
-    # 保存到storage的路径
-    target_path = os.path.join(default_storage.base_location, local_path)
-    target_dir = os.path.dirname(target_path)
-    if not os.path.isdir(target_dir):
-        make_dirs(target_dir, exist_ok=True)
-
-    ok, err = storage.download(session_path, target_path)
-    if not ok:
-        msg = "Failed download replay file: {}".format(err)
-        logger.error(msg)
-        return None, msg
-    url = default_storage.url(local_path)
-    return local_path, url
-
-
-def get_session_replay_url(session):
-    local_path, url = find_session_replay_local(session)
-    if local_path is None:
-        local_path, url = download_session_replay(session)
-    return local_path, url
-