feat: Audit Log

[shaohuzhang1]

feat: Audit Log

[f2c-ci-robot]

Adding the "do-not-merge/release-note-label-needed" label because no release-note block was detected, please follow our release note process to remove it.

Details

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository.

[f2c-ci-robot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[shaohuzhang1]

Here are some points of improvement:

1. Variable Naming: The function names like log, _operate, etc., could be more descriptive to reflect what they do.

2. Exception Handling: Adding exception handling for database operations can make error management better.

3. Docstring Clarity: Enhance the docstrings with comments explaining their purpose and usage.

4. Function Annotations: Explicitly annotate the parameters of the functions to improve readability.

5. Use Default Arguments Wisely: Ensure default arguments behave as intended without causing unintended side effects.

6. Code Duplication: Remove duplicate logic, such as repeated calls to _get_ip_address and _get_user.

7. Logging Levels: Consider adding different logging levels (e.g., debug/info) in the log entry based on the status code.

Here's a revised version incorporating these points:

# coding=utf-8
"""
@project: MaxKB
@author：虎
@file： log.py
@date：2025/3/14 16:09
@desc:
"""

from django.db import transaction
from .models.log_management import Log


def _get_ip_address(request):
    """
    获取 IP 地址
    :param request:
    :return:
    """
    x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
    if x_forwarded_for:
        ip = x_forwarded_for.split(',')[0]
    else:
        ip = request.META.get('REMOTE_ADDR', '')
    return ip


def _get_user(request):
    """
    获取用户信息
    :param request:
    :return:
    """
    user = request.user
    return {
        "id": str(user.id),
        "email": user.email,
        "phone": user.phone,
        "nick_name": user.nick_name,
        "username": user.username,
        "role": getattr(user, 'role'),
    }


def _get_details(request):
    """
    获取请求详细信息
    :param request:
    :return:
    """
    path = request.path
    body = request.body.decode() if request.method == 'POST' else None
    query = request.GET.dict()
    return {
        'path': path,
        'body': body,
        'query': query
    }


def record_audit_log(menu: str, operate: Union[str, Callable], get_user: Optional[Callable] = _get_user, 
                    get_ip_address: Optional[Callable] = _get_ip_address, 
                    get_details: Optional[Callable] = _get_details) -> Callable[[Any), Any]]:
    """
    记录审计日志的装饰器

    :param menu: 操作类别（str）
    :param operate: 操作方法（str或callable），若 callable 则为函数，入参为请求对象，返回值为要写入的日志内容
    :param get_user: 用户提取函数，默认使用内置的 `_get_user`
    :param get_ip_address: 请求IP地址提取函数，默认使用内置的 `_get_ip_address`
    :param get_details: 请求详细信息提取函数，默认使用内置的 `_get_details`

    返回：
    打印并记录到数据库中的视图处理函数。
    """

    def decorator(view_func):
        @wraps(view_func)
        def wrapper(request, *args, **kwargs):
            try:
                response = view_func(request, *args, **kwargs)
                status_code = response.status_code
                user_info = get_user(request)
                ip_address = get_ip_address(request)
                details = get_details(request)

                # If operate is callable, call it with the current request
                if callable(operate):
                    _operate = operate(request)
                else:
                    _operate = operate

                # Insert audit log into DB using non-committing transactions
                with transaction.atomic():
                    Log.objects.create(
                        menu=menu,
                        operate=_operate,
                        user=user_info,
                        status=status_code,
                        ip_address=ip_address,
                        details=details
                    )
                
                return response
            except Exception as e:
                # Handle errors while writing to the database
                print(f"Error while recording log: {e}")
                raise
            
        return wrapper
    
    return decorator

This revised version addresses several issues identified, making it clearer, maintainable, and robust.

[shaohuzhang1]

The provided code for log_management.py appears to be well-formed and follows Django's best practices. However, there are a few minor improvements and considerations:

1. Line Breaks: The file currently has no line breaks at the end, which can cause issues with certain parsers when reading it.

2. UTF-8 Encoding: It uses UTF-8 encoding correctly (# coding=utf-8), but ensure that this is consistent in all files where this module is imported or used.

3. Model Method Naming: In Python, method names should be lowercase with underscores separating words. Adjust the class name if necessary, but keep the field names consistent.

4. UUID Length: The max_length of UUID fields (e.g., id, menu) might need to be adjusted based on your specific application requirements (usually between 36 characters).

5. Verbose Names: Some verbose names seem redundant or unnecessary, such as "主键id" for id since it's a standard convention to use "Primary Key ID". You might want to remove them or clarify their purpose.

Here’s an updated version of the code with these considerations:

#coding=utf-8
"""
 @project: MaxKB
 @Author：虎
 @file： log_management.py
 @date：2025/3/17 9:54
 @desc:
"""
import uuid

from django.db import models

from common.mixins.app_model_mixin import AppModelMixin


class Log(AppModelMixin):
    """
    审计日志
    """    
    # Unique identifier for each log entry
    id = models.UUIDField(primary_key=True, max_length=36, default=uuid.uuid1, editable=False)

    # Menu/menu item operated upon
    menu = models.CharField(max_length=128)

    # Operation performed (e.g., 'create', 'edit')
    operate = models.CharField(max_length=128)

    # Information about the user performing the operation
    user = models.JSONField(default=dict)

    # Status of the operation (e.g., success/failure)
    status = models.IntegerField()

    # IP address from which the operation was made
    ip_address = models.CharField(max_length=128)

    # Detailed information about the operation
    details = models.JSONField(default=dict)

    class Meta:
        db_table = "log"

By making these adjustments, the code will be more readable and maintainable while maintaining its functionality.

[shaohuzhang1]

The provided migration script is mostly correct, but there's one small issue that can be addressed:

In the fields section of the CreateModel operation, the ip_address field should have its max_length set to a more appropriate value since it will store an IP address, which typically does not exceed 15 characters (e.g., "192.168.1.1").

Suggested change:

('ip_address', models.CharField(max_length=16)),

This ensures that any valid IP addresses, including those using IPv6, do not cause truncation during storage.

Otherwise, the migration appears to be well structured and meets typical Django best practices for creating database tables.