File size: 9,208 Bytes
020c337 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 | import json
import os
import sys
import time
from typing import Tuple, Union, Optional, Dict
from .mods import PUSH_DATA_PATH, TaskTemplateConfig, TaskConfig
import traceback
from .base_task import BaseTask, BaseTaskViewMsg
from .site_push import web_info_data
from .util import read_file, DB, GET_CLASS, write_file, debug_log, to_dict_obj, random_string, get_webserver
class WEBLogTask(BaseTask):
CROM_CONFIG_FILE = "/www/server/panel/data/cron_task_analysis.json"
def __init__(self):
super().__init__()
self.source_name: str = 'web_log_scan'
self.title: str = 'Web日志扫描告警' # 这个是告警任务的标题(根据实际情况改变)
self.template_name: str = 'Web日志扫描告警' # 这个告警模板的标题(不会改变)
@staticmethod
def get_site_log_file(site_name: str) -> str:
if "/www/server/panel/class" not in sys.path:
sys.path.insert(0, "/www/server/panel/class")
from logsModel.siteModel import main as SiteModel
res = SiteModel().get_site_log_file(to_dict_obj({"siteName": site_name}))
if "log_file" in res and res["log_file"]:
return res["log_file"]
return ""
def check_task_data(self, task_data: dict) -> Union[dict, str]:
"""
检查设置的告警参数（是否合理）
@param task_data: 传入的告警参数，提前会经过默认值处理（即没有的字段添加默认值）
@return: 当检查无误时，返回一个 dict 当做后续的添加和修改的数据，
当检查有误时， 直接返回错误信息的字符串
"""
if "interval" in task_data:
task_data.pop("interval")
site_name = task_data.get("site_name", "")
if not site_name:
return "请配置站点名称"
items, _ = web_info_data(all_type=True)
for item in items:
if item["value"] == site_name:
break
else:
return "站点不存在"
task_data["log_path"] = self.get_site_log_file(site_name)
if not os.path.exists(task_data["log_path"]):
return "站点日志文件不存在"
return task_data
def get_keyword(self, task_data: dict) -> str:
"""
返回一个关键字，用于后续查询或执行任务时使用， 例如：防篡改告警，可以根据其规则id生成一个关键字，
后续通过规则id和来源tamper 查询并使用
@param task_data: 通过check_args后生成的告警参数字典
@return: 返回一个关键词字符串
"""
return "web_log_scan_{}".format(task_data["log_path"])
def get_title(self, task_data: dict) -> str:
"""
返回一个标题
@param task_data: 通过check_args后生成的告警参数字典
@return: 返回一个关键词字符串
"""
return "网站【{}】Web日志扫描提醒".format(task_data["site_name"])
def get_push_data(self, task_id: str, task_data: dict) -> Optional[dict]:
"""
判断这个任务是否需要返送
@param task_id: 任务id
@param task_data: 任务的告警参数
@return: 如果触发了告警，返回一个dict的原文，作为告警信息，否则应当返回None表示未触发
返回之中应当包含一个 msg_list 的键（值为List[str]类型），将主要的信息返回
用于以下信息的自动序列化包含[dingding, feishu, mail, weixin, web_hook]
短信和微信公众号由于长度问题，必须每个任务手动实现
"""
return None
def filter_template(self, template: dict) -> Optional[dict]:
"""
过滤 和 更改模板中的信息, 返回空表是当前无法设置该任务
@param template: 任务的模板信息
@return:
"""
items, _ = web_info_data(all_type=True)
template["field"][0]["items"].extend(items)
return template
@classmethod
def all_web_log_scan(cls) -> Dict[str, str]:
res = {}
for i in TaskConfig().get_by_source("web_log_scan"):
if i["status"] is True:
if "log_path" not in i["task_data"]:
i["task_data"]["log_path"] = cls.get_site_log_file(i["task_data"]["site_name"])
res[i["task_data"]["log_path"]] = i["task_data"]["site_name"]
return res
@classmethod
def set_path_to_config(cls, path: str, status: bool):
try:
conf = json.loads(read_file(cls.CROM_CONFIG_FILE))
except:
conf = {}
if path in conf and not status: # 移除配置
conf.pop(path)
elif path not in conf and status: # 新增配置
conf[path] = {"status": 1, "cycle": '1', "path": path}
if len(conf) > 1:
cls.add_crontab()
else:
cls.remove_crontab()
write_file(cls.CROM_CONFIG_FILE, json.dumps(conf))
def task_config_create_hook(self, task: dict) -> None:
self.set_path_to_config(task["task_data"]["log_path"], task["status"])
def task_config_update_hook(self, task: dict) -> None:
self.set_path_to_config(task["task_data"]["log_path"], task["status"])
def task_config_remove_hook(self, task: dict) -> None:
self.set_path_to_config(task["task_data"]["log_path"], False)
@staticmethod
def add_crontab():
"""
@name 构造日志切割任务
"""
cron_name = '[勿删]web日志定期检测服务'
if not DB('crontab').where('name=?', (cron_name,)).count():
cmd = '{} /www/server/panel/script/cron_log_analysis.py'.format(
"/www/server/panel/pyenv/bin/python3.7"
)
args = {
"name": cron_name,
"type": 'day',
"where1": '',
"hour": '11',
"minute": '50',
"sName": "",
"sType": 'toShell',
"notice": '0',
"notice_channel": '',
"save": '',
"save_local": '1',
"backupTo": '',
"sBody": cmd,
"urladdress": ''
}
if "/www/server/panel/class" not in sys.path:
sys.path.insert(0, "/www/server/panel/class")
import crontab
res = crontab.crontab().AddCrontab(args)
if res and "id" in res.keys():
return True
return False
return True
@staticmethod
def remove_crontab():
"""
@name 删除项目定时任务
@auther hezhihong<2022-10-31>
@return
"""
try:
cron_name = '[勿删]web日志定期检测服务'
if "/www/server/panel/class" not in sys.path:
sys.path.insert(0, "/www/server/panel/class")
import crontab
p = crontab.crontab()
cron_id = DB('crontab').where("name=?", (cron_name,)).getField('id')
args = {"id": cron_id}
p.DelCrontab(args)
return True
except:
return False
def load_web_log_template():
from .mods import load_task_template_by_config
load_task_template_by_config(
[{
"id": "110",
"ver": "1",
"used": True,
"source": "web_log_scan",
"title": "Web日志扫描告警",
"load_cls": {
"load_type": "path",
"cls_path": "mod.base.push_mod.web_log_push",
"name": "WEBLogTask"
},
"template": {
"field": [
{
"attr": "site_name",
"name": "站点名称",
"type": "select",
"default": "",
"unit": "",
"suffix": "",
"items": []
}
],
"sorted": [
[
"site_name"
],
],
},
"default": {},
"advanced_default": {},
"send_type_list": [
"dingding",
"feishu",
"mail",
"weixin",
"webhook"
],
"unique": False,
"tags":["site"],
"description": "定期扫描网站的Nginx或Apache访问日志，检查潜在的风险，（如：SQL 注入、XSS 攻击、命令执行等）并通知管理人员。"
}]
)
class ViewMsgFormat(BaseTaskViewMsg):
def __init__(self):
self.web = get_webserver()
def get_msg(self, task: dict) -> Optional[str]:
if task["template_id"] == "110":
return "定期扫描网站【{}】的{}访问日志并报告PHP攻击、恶意扫描、SQL注入、XSS攻击的情况".format(task["task_data"]["site_name"], self.web)
return None
WEBLogTask.VIEW_MSG = ViewMsgFormat |