jumpserver/apps/audits/serializers.py

# -*- coding: utf-8 -*-
#
from django.utils.translation import ugettext_lazy as _
from rest_framework import serializers

from audits.backends.db import OperateLogStore
from common.serializers.fields import LabeledChoiceField
from common.utils import reverse
from common.utils.timezone import as_current_tz
from ops.models.job import JobAuditLog
from ops.serializers.job import JobExecutionSerializer
from terminal.models import Session
from . import models
from .const import (
    ActionChoices, OperateChoices,
    MFAChoices, LoginStatusChoices,
    LoginTypeChoices, ActivityChoices,
)


class JobAuditLogSerializer(JobExecutionSerializer):
    class Meta:
        model = JobAuditLog
        read_only_fields = [
            "id", "material", "time_cost", 'date_start',
            'date_finished', 'date_created',
            'is_finished', 'is_success', 'created_by',
            'task_id'
        ]
        fields = read_only_fields + []


class FTPLogSerializer(serializers.ModelSerializer):
    operate = LabeledChoiceField(choices=OperateChoices.choices, label=_("Operate"))

    class Meta:
        model = models.FTPLog
        fields_mini = ["id"]
        fields_small = fields_mini + [
            "user", "remote_addr", "asset", "account",
            "org_id", "operate", "filename", "is_success",
            "date_start",
        ]
        fields = fields_small


class UserLoginLogSerializer(serializers.ModelSerializer):
    mfa = LabeledChoiceField(choices=MFAChoices.choices, label=_("MFA"))
    type = LabeledChoiceField(choices=LoginTypeChoices.choices, label=_("Type"))
    status = LabeledChoiceField(choices=LoginStatusChoices.choices, label=_("Status"))

    class Meta:
        model = models.UserLoginLog
        fields_mini = ["id"]
        fields_small = fields_mini + [
            "username", "type", "ip",
            "city", "user_agent", "mfa",
            "reason", "reason_display",
            "backend", "backend_display",
            "status", "datetime",
        ]
        fields = fields_small
        extra_kwargs = {
            "user_agent": {"label": _("User agent")},
            "reason_display": {"label": _("Reason display")},
            "backend_display": {"label": _("Authentication backend")},
        }


class OperateLogActionDetailSerializer(serializers.ModelSerializer):
    class Meta:
        model = models.OperateLog
        fields = ('diff',)

    def to_representation(self, instance):
        data = super().to_representation(instance)
        diff = OperateLogStore.convert_diff_friendly(data['diff'])
        data['diff'] = diff
        return data


class OperateLogSerializer(serializers.ModelSerializer):
    action = LabeledChoiceField(choices=ActionChoices.choices, label=_("Action"))

    class Meta:
        model = models.OperateLog
        fields_mini = ["id"]
        fields_small = fields_mini + [
            "user", "action", "resource_type",
            "resource_type_display", "resource",
            "remote_addr", "datetime", "org_id",
        ]
        fields = fields_small
        extra_kwargs = {"resource_type_display": {"label": _("Resource Type")}}


class PasswordChangeLogSerializer(serializers.ModelSerializer):
    class Meta:
        model = models.PasswordChangeLog
        fields = ("id", "user", "change_by", "remote_addr", "datetime")


class SessionAuditSerializer(serializers.ModelSerializer):
    class Meta:
        model = Session
        fields = "__all__"


class ActivityOperatorLogSerializer(serializers.Serializer):
    timestamp = serializers.SerializerMethodField()
    detail_url = serializers.SerializerMethodField()
    content = serializers.SerializerMethodField()

    @staticmethod
    def get_timestamp(obj):
        return as_current_tz(obj['datetime']).strftime('%Y-%m-%d %H:%M:%S')

    @staticmethod
    def get_content(obj):
        if not obj['r_detail']:
            action = obj['r_action'].replace('_', ' ').capitalize()
            ctn = _('User {} {} this resource.').format(obj['r_user'], _(action))
        else:
            ctn = obj['r_detail']
        return ctn

    @staticmethod
    def get_detail_url(obj):
        detail_url = ''
        detail_id, obj_type = obj['r_detail_id'], obj['r_type']
        if not detail_id:
            return detail_url

        if obj_type == ActivityChoices.operate_log:
            detail_url = reverse(
                view_name='audits:operate-log-detail',
                kwargs={'pk': obj['id']},
                api_to_ui=True, is_audit=True
            )
        elif obj_type == ActivityChoices.task:
            detail_url = reverse(
                'ops:celery-task-log', kwargs={'pk': detail_id}
            )
        elif obj_type == ActivityChoices.login_log:
            detail_url = '%s?id=%s' % (
                reverse('api-audits:login-log-list', api_to_ui=True, is_audit=True),
                detail_id
            )
        return detail_url