Python中如何在Django框架下实现日志审查功能
现有的系统下,要增加一个日志审计。
现有的环境: Django+Mongodb 。
然后想直接用 logging 来处理日志,但是怎么入库呢?
看到网上也有人讨论 signals 来处理,这个块不是很熟,想问一下,只处理 views 可以吗,因为现有的环境没有 models 的处理。 谢谢
Python中如何在Django框架下实现日志审查功能
5 回复
自定义信号,可以处理 views
没试过 logging 日志入库
在Django里搞日志审查,核心就是配置好LOGGING,然后用logging模块记录关键操作。下面是一个可以直接用的配置示例。
首先,在你的settings.py里配置日志。这个配置会把INFO及以上级别的日志同时输出到控制台和一个按日期滚动的文件里,并且给管理员发邮件。
# settings.py
import os
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '{levelname} {asctime} {module} {process:d} {thread:d} {message}',
'style': '{',
},
'simple': {
'format': '{levelname} {message}',
'style': '{',
},
'audit': {
'format': '{asctime} - {user} - {ip} - {action} - {status}',
},
},
'handlers': {
'console': {
'level': 'INFO',
'class': 'logging.StreamHandler',
'formatter': 'simple'
},
'file': {
'level': 'INFO',
'class': 'logging.handlers.TimedRotatingFileHandler',
'filename': os.path.join(BASE_DIR, 'logs', 'django.log'),
'when': 'midnight',
'backupCount': 30,
'formatter': 'verbose',
},
'audit_file': {
'level': 'INFO',
'class': 'logging.handlers.TimedRotatingFileHandler',
'filename': os.path.join(BASE_DIR, 'logs', 'audit.log'),
'when': 'midnight',
'backupCount': 90,
'formatter': 'audit',
},
'mail_admins': {
'level': 'ERROR',
'class': 'django.utils.log.AdminEmailHandler',
}
},
'loggers': {
'django': {
'handlers': ['console', 'file', 'mail_admins'],
'level': 'INFO',
'propagate': True,
},
'django.request': {
'handlers': ['console', 'file', 'mail_admins'],
'level': 'ERROR',
'propagate': False,
},
'myapp.audit': {
'handlers': ['audit_file'],
'level': 'INFO',
'propagate': False,
},
},
}
然后,在你需要记录审计日志的地方(比如视图里),用自定义的格式记录。这里的关键是使用logger.info()并传入一个包含审计信息的字典。
# views.py
import logging
from django.contrib.auth.decorators import login_required
from django.utils.decorators import method_decorator
from django.views import View
from django.http import JsonResponse
# 获取我们配置的审计日志记录器
audit_logger = logging.getLogger('myapp.audit')
class SensitiveActionView(View):
@method_decorator(login_required)
def post(self, request):
user = request.user
action = 'data_export'
client_ip = request.META.get('REMOTE_ADDR', '')
try:
# 这里是你的核心业务逻辑,比如导出数据
# ... 你的代码 ...
# 记录成功的审计日志
audit_logger.info('', extra={
'user': user.username,
'ip': client_ip,
'action': action,
'status': 'SUCCESS'
})
return JsonResponse({'status': 'ok'})
except Exception as e:
# 记录失败的审计日志
audit_logger.error('', extra={
'user': user.username,
'ip': client_ip,
'action': action,
'status': f'FAILED: {str(e)}'
})
return JsonResponse({'status': 'error', 'message': str(e)}, status=500)
最后,记得在你的项目根目录下创建logs文件夹,不然文件会写不进去。审计日志会单独记录在audit.log里,格式清晰,方便你后续用脚本或者ELK之类的工具来分析。
总结一下就是配好LOGGING,在关键操作点用特定格式记录日志。
光是 log 也看不到具体的操作记录吧
格式化一个对象,异步丢到 celery 里面去入库吧。
+1

