mirror of
https://github.com/aaPanel/aaPanel.git
synced 2026-08-17 21:25:47 +02:00
2074 lines
79 KiB
Python
2074 lines
79 KiB
Python
# coding: utf-8
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
from collections import deque
|
|
|
|
import public
|
|
from safeModelV2.base import safeBase
|
|
|
|
|
|
class main(safeBase):
|
|
"""Security overview read model.
|
|
|
|
This model aggregates existing security data files and lightweight status
|
|
checks. It must not start scans or change security settings.
|
|
"""
|
|
|
|
EVENT_LIMIT_MAX = 100
|
|
|
|
SEVERITY_ORDER = {
|
|
"info": 0,
|
|
"low": 1,
|
|
"medium": 2,
|
|
"high": 3,
|
|
"critical": 4,
|
|
}
|
|
|
|
SEVERITY_LABELS = {
|
|
"info": "Info",
|
|
"low": "Low",
|
|
"medium": "Medium",
|
|
"high": "High",
|
|
"critical": "Critical",
|
|
}
|
|
|
|
# 安全/插件 跳转路由
|
|
MODULE_ROUTES = {
|
|
"firewall": "/safe/firewall",
|
|
"ssh": "/safe/ssh",
|
|
"fail2ban": "/plugin/fail2ban",
|
|
"tamper_core": "/plugin/tamper_core",
|
|
"btwaf": "/waf/overview",
|
|
}
|
|
# 安全功能 跳转路由
|
|
SCAN_CARD_ROUTES = {
|
|
"website_vulnerability": "/security/scanner",
|
|
"malware_detection": "/security/malware",
|
|
"security_risk": "/security/risk",
|
|
"baseline": "/security/baseline",
|
|
}
|
|
|
|
|
|
SCAN_CARD_SOURCE_APIS = {
|
|
"website_vulnerability": "/v2/scanning?action=get_vuln_info",
|
|
"malware_detection": "/v2/safecloud?action=get_webshell_result",
|
|
"security_risk": "/v2/warning?action=get_list",
|
|
"baseline": "/v2/safe_detect?action=get_safe_count",
|
|
}
|
|
|
|
# 企业防篡改插件的防御篡改类型
|
|
TAMPER_INTERCEPT_KEYS = [
|
|
"create", "modify", "unlink", "rename", "mkdir",
|
|
"rmdir", "chmod", "chown", "link", "warning_msg",
|
|
]
|
|
|
|
# waf插件的防御类型
|
|
WAF_INTERCEPT_KEYS = [
|
|
"cc", "xss", "sql", "user_agent", "cookie", "scan", "upload",
|
|
"path_php", "download", "smart_cc", "file", "php", "other",
|
|
"file_import", "path", "url_tell", "url_rule",
|
|
]
|
|
|
|
# 插件
|
|
PLUGIN_ACTION_SOURCES = set(["fail2ban", "tamper_core", "btwaf"])
|
|
|
|
# 概览页扫描项 后续加扫描项在此追加
|
|
SCAN_TASKS = {
|
|
"security_risk": {
|
|
"title": "Security risk scan",
|
|
"source_type": "panel",
|
|
"api": "/v2/warning",
|
|
"params": {"action": "get_list", "force": 1},
|
|
},
|
|
}
|
|
|
|
def __init__(self):
|
|
safeBase.__init__(self)
|
|
self.panel_path = public.get_panel_path()
|
|
|
|
def get_overview(self, get):
|
|
"""
|
|
@name Get security overview for the new security overview page.
|
|
@param get.event_limit int optional, default 10, max 100
|
|
@param get.include_events int optional, default 1
|
|
@param get.include_modules int optional, default 1
|
|
"""
|
|
try:
|
|
event_limit = self._get_int(get, "event_limit", 10, 1, self.EVENT_LIMIT_MAX)
|
|
include_events = self._get_bool(get, "include_events", True)
|
|
include_modules = self._get_bool(get, "include_modules", True)
|
|
|
|
scan_cards = self._get_scan_cards()
|
|
protection = self._get_protection_status()
|
|
health = self._build_health(scan_cards)
|
|
actions = self._build_recommended_actions(scan_cards, protection["items"])
|
|
events = self._get_events_data(limit=event_limit) if include_events else []
|
|
modules = self._get_modules(scan_cards, protection["items"]) if include_modules else []
|
|
|
|
scan_has_data = any(card["has_data"] for card in scan_cards)
|
|
result = {
|
|
"server_time": int(time.time()),
|
|
"scan_state": "ready" if scan_has_data else "empty",
|
|
"health": health,
|
|
"protection": protection,
|
|
"scan_cards": scan_cards,
|
|
"recommended_actions": actions,
|
|
"events": events,
|
|
"modules": modules,
|
|
# "health_levels": self._health_levels(),
|
|
# "source_status": self._source_status(scan_cards),
|
|
# "scan_entry": self._scan_entry(),
|
|
}
|
|
return public.return_message(0, 0, result)
|
|
except Exception as ex:
|
|
public.print_log("security overview get_overview error: {}".format(ex))
|
|
return public.return_message(-1, 0, str(ex))
|
|
|
|
def get_events(self, get):
|
|
"""
|
|
@name Get aggregated security events.
|
|
@param get.p int optional, default 1
|
|
@param get.limit int optional, default 20, max 100
|
|
@param get.days int optional, default 7, 0 means no day filter
|
|
@param get.source string optional
|
|
@param get.type string optional
|
|
@param get.level string optional, one of info/low/medium/high/critical
|
|
@param get.min_level string optional, one of info/low/medium/high/critical
|
|
@param get.keyword string optional
|
|
"""
|
|
try:
|
|
p = self._get_int(get, "p", 1, 1, 1000000)
|
|
limit = self._get_int(get, "limit", 20, 1, self.EVENT_LIMIT_MAX)
|
|
days = self._get_int(get, "days", 7, 0, 3660)
|
|
source = self._get_arg(get, "source", "")
|
|
# event_type = self._get_arg(get, "type", "")
|
|
level = self._normalize_level(self._get_arg(get, "level", ""))[0]
|
|
min_level = self._normalize_level(self._get_arg(get, "min_level", ""))[0]
|
|
keyword = str(self._get_arg(get, "keyword", "") or "").lower()
|
|
|
|
events = self._get_events_data(limit=500)
|
|
if days > 0:
|
|
start_time = int(time.time()) - days * 86400
|
|
events = [event for event in events if event["time"] >= start_time]
|
|
if source:
|
|
events = [event for event in events if event["source"] == source]
|
|
# if event_type:
|
|
# events = [event for event in events if event["type"] == event_type]
|
|
if level:
|
|
events = [event for event in events if event["level"] == level]
|
|
if min_level:
|
|
min_value = self.SEVERITY_ORDER[min_level]
|
|
events = [
|
|
event for event in events
|
|
if self.SEVERITY_ORDER.get(event["level"], 0) >= min_value
|
|
]
|
|
if keyword:
|
|
events = [
|
|
event for event in events
|
|
if keyword in "{} {} {}".format(
|
|
event.get("title", ""),
|
|
event.get("message", ""),
|
|
event.get("meta", ""),
|
|
).lower()
|
|
]
|
|
|
|
total = len(events)
|
|
start = (p - 1) * limit
|
|
data = events[start:start + limit]
|
|
public.set_module_logs('safe_overview', 'get_events')
|
|
return public.return_message(0, 0, {
|
|
"p": p,
|
|
"limit": limit,
|
|
"total": total,
|
|
"data": data,
|
|
# "stats": self._event_stats(events),
|
|
})
|
|
except Exception as ex:
|
|
public.print_log("security overview get_events error: {}".format(ex))
|
|
return public.return_message(-1, 0, str(ex))
|
|
|
|
def get_modules(self, get):
|
|
"""
|
|
@name Get security module and plugin status list.
|
|
"""
|
|
try:
|
|
scan_cards = self._get_scan_cards()
|
|
protection = self._get_protection_status()
|
|
return public.return_message(0, 0, self._get_modules(scan_cards, protection["items"]))
|
|
except Exception as ex:
|
|
public.print_log("security overview get_modules error: {}".format(ex))
|
|
return public.return_message(-1, 0, str(ex))
|
|
|
|
def scan_now(self, get):
|
|
"""
|
|
@name Run security overview scan tasks.
|
|
@param get.scan_types array/string optional, default security_risk
|
|
@param get.force int/bool optional, default 1
|
|
"""
|
|
try:
|
|
scan_types = self._get_scan_types(get)
|
|
force = self._get_bool(get, "force", True)
|
|
|
|
for scan_type in scan_types:
|
|
self._run_scan_task(scan_type, force=force)
|
|
public.set_module_logs('safe_overview', 'scan_now')
|
|
return public.return_message(0, 0, public.lang('Scan completed'))
|
|
except Exception as ex:
|
|
public.print_log("security overview scan_now error: {}".format(ex))
|
|
return public.return_message(-1, 0, str(ex))
|
|
|
|
def toggle_plugin(self, get):
|
|
"""
|
|
@name Toggle security overview plugin status.
|
|
@param get.plugin_name string required, fail2ban/btwaf/tamper_core
|
|
@param get.type string optional, fail2ban start/stop
|
|
@param get.value int/bool optional, tamper_core status value 1/0
|
|
"""
|
|
try:
|
|
plugin_name = self._get_toggle_plugin_name(get)
|
|
if plugin_name not in self.PLUGIN_ACTION_SOURCES:
|
|
return public.return_message(-1, 0, "Unsupported plugin")
|
|
|
|
if plugin_name == "fail2ban":
|
|
result, action, target_enabled = self._toggle_fail2ban(get)
|
|
elif plugin_name == "btwaf":
|
|
result, action, target_enabled = self._toggle_btwaf(get)
|
|
else:
|
|
result, action, target_enabled = self._toggle_tamper_core(get)
|
|
|
|
success = self._plugin_result_success(result)
|
|
if success:
|
|
return public.return_message(0 , 0, public.lang("Set successfully!"))
|
|
else:
|
|
return public.return_message(-1, 0, public.lang("Set failed"))
|
|
|
|
return public.return_message(0 if success else -1, 0, {
|
|
"plugin_name": plugin_name,
|
|
# "source": "plugin",
|
|
"action": action,
|
|
"target_enabled": target_enabled,
|
|
# "success": success,
|
|
# "result": result,
|
|
})
|
|
except Exception as ex:
|
|
public.print_log("security overview toggle_plugin error: {}".format(ex))
|
|
return public.return_message(-1, 0, str(ex))
|
|
|
|
def get_health_levels(self, get):
|
|
"""
|
|
@name Get static health level definitions.
|
|
"""
|
|
return public.return_message(0, 0, self._health_levels())
|
|
|
|
# ------------------------------
|
|
# Plugin toggles
|
|
# ------------------------------
|
|
|
|
def _get_toggle_plugin_name(self, get):
|
|
plugin_name = str(self._get_first_arg(get, ["plugin_name"], "") or "").strip().lower()
|
|
return plugin_name
|
|
|
|
def _toggle_fail2ban(self, get):
|
|
action = self._normalize_service_action(self._get_first_arg(get, ["type"], ""))
|
|
if action not in ["start", "stop"]:
|
|
raise ValueError("Fail2Ban requires type=start or type=stop")
|
|
|
|
self._record_toggle_log("fail2ban")
|
|
result = public.run_plugin(
|
|
"fail2ban",
|
|
"set_fail2ban_status",
|
|
public.to_dict_obj({"type": action})
|
|
)
|
|
return result, action, action == "start"
|
|
|
|
def _toggle_btwaf(self, get):
|
|
plugin_dir = self._plugin_path("btwaf")
|
|
raw_desired = self._get_first_arg(get, ["value", "open", "enabled"], None)
|
|
desired = self._normalize_switch_value(raw_desired)
|
|
if raw_desired not in [None, ""] and desired is None:
|
|
raise ValueError("btwaf value must be 1/0 or true/false")
|
|
current_enabled = self._get_btwaf_global_open(plugin_dir)
|
|
action = "toggle"
|
|
|
|
self._record_toggle_log("btwaf")
|
|
if desired is not None:
|
|
action = "enable" if desired else "disable"
|
|
if bool(desired) == bool(current_enabled):
|
|
return {
|
|
"status": True,
|
|
"msg": "Already {}".format("enabled" if desired else "disabled"),
|
|
}, action, bool(desired)
|
|
|
|
result = public.run_plugin("btwaf", "set_open", public.to_dict_obj({}))
|
|
target_enabled = bool(desired) if desired is not None else not bool(current_enabled)
|
|
return result, action, target_enabled
|
|
|
|
def _toggle_tamper_core(self, get):
|
|
raw_value = self._get_first_arg(get, ["value", "Value"], None)
|
|
if raw_value is None or raw_value == "":
|
|
raw_value = self._get_first_arg(get, ["action", "type", "enabled", "status"], None)
|
|
value = self._normalize_switch_value(raw_value)
|
|
if value is None:
|
|
raise ValueError("tamper_core requires value=1 or value=0")
|
|
|
|
self._record_toggle_log("tamper_core")
|
|
result = public.run_plugin(
|
|
"tamper_core",
|
|
"modify_global_config",
|
|
public.to_dict_obj({"key": "status", "value": value})
|
|
)
|
|
return result, "enable" if value else "disable", bool(value)
|
|
|
|
def _record_toggle_log(self, plugin_name):
|
|
"""
|
|
记录埋点
|
|
"""
|
|
actions = {
|
|
"fail2ban": "toggle_fail2ban",
|
|
"btwaf": "toggle_btwaf",
|
|
"tamper_core": "toggle_tamper_core",
|
|
}
|
|
action = actions.get(plugin_name)
|
|
if not action:
|
|
return
|
|
|
|
public.set_module_logs('safe_overview', action)
|
|
|
|
|
|
def _normalize_service_action(self, value):
|
|
text = "" if value is None else str(value).strip().lower()
|
|
if text in ["start", "enable", "enabled", "open", "on", "1", "true", "yes"]:
|
|
return "start"
|
|
if text in ["stop", "disable", "disabled", "close", "closed", "off", "0", "false", "no"]:
|
|
return "stop"
|
|
return text
|
|
|
|
def _normalize_switch_value(self, value):
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, bool):
|
|
return 1 if value else 0
|
|
if isinstance(value, int):
|
|
return 1 if value != 0 else 0
|
|
text = str(value).strip().lower()
|
|
if not text:
|
|
return None
|
|
if text in ["1", "true", "yes", "on", "open", "start", "enable", "enabled"]:
|
|
return 1
|
|
if text in ["0", "false", "no", "off", "close", "closed", "stop", "disable", "disabled"]:
|
|
return 0
|
|
return None
|
|
|
|
def _plugin_result_success(self, result):
|
|
if isinstance(result, bool):
|
|
return result
|
|
if isinstance(result, dict):
|
|
if "status" in result:
|
|
status = result.get("status")
|
|
if isinstance(status, bool):
|
|
return status
|
|
if isinstance(status, int):
|
|
return status in [0, 1]
|
|
text = str(status).strip().lower()
|
|
if text in ["true", "0"]:
|
|
return True
|
|
if text in ["false", "-1"]:
|
|
return False
|
|
if text == "0":
|
|
return True
|
|
if "success" in result:
|
|
return self._to_bool(result.get("success"), True)
|
|
return True
|
|
|
|
# ------------------------------
|
|
# Scan runners
|
|
# ------------------------------
|
|
|
|
def _get_scan_types(self, get):
|
|
value = self._get_arg(get, "scan_types", None)
|
|
if value is None or value == "":
|
|
value = self._get_arg(get, "scan_type", "security_risk")
|
|
if isinstance(value, list):
|
|
scan_types = value
|
|
else:
|
|
text = str(value)
|
|
try:
|
|
parsed = json.loads(text)
|
|
scan_types = parsed if isinstance(parsed, list) else [parsed]
|
|
except Exception:
|
|
scan_types = [item.strip() for item in text.split(",")]
|
|
|
|
result = []
|
|
for scan_type in scan_types:
|
|
scan_type = str(scan_type).strip()
|
|
if scan_type in self.SCAN_TASKS and scan_type not in result:
|
|
result.append(scan_type)
|
|
return result or ["security_risk"]
|
|
|
|
def _run_scan_task(self, scan_type, force=True):
|
|
task = self.SCAN_TASKS.get(scan_type, {})
|
|
started_at = int(time.time())
|
|
base = {
|
|
"id": scan_type,
|
|
"title": task.get("title", scan_type),
|
|
"source_type": task.get("source_type", "panel"),
|
|
"api": task.get("api", ""),
|
|
"params": task.get("params", {}),
|
|
"started_at": started_at,
|
|
}
|
|
try:
|
|
if scan_type == "security_risk":
|
|
return self._run_security_risk_scan(base, force=force)
|
|
return self._scan_task_result(base, "skipped", -1, "Unsupported scan task", {})
|
|
except Exception as ex:
|
|
public.print_log("security overview scan task {} error: {}".format(scan_type, ex))
|
|
return self._scan_task_result(base, "failed", -1, str(ex), {})
|
|
|
|
def _run_security_risk_scan(self, base, force=True):
|
|
import panel_warning_v2
|
|
args = public.to_dict_obj({
|
|
"force": 1 if force else 0,
|
|
})
|
|
started = time.time()
|
|
result = panel_warning_v2.panelWarning().get_list(args)
|
|
message = self._unwrap_message(result, {})
|
|
code = self._safe_int(result.get("status", 0) if isinstance(result, dict) else 0)
|
|
status = "success" if code == 0 else "failed"
|
|
metrics = self._security_risk_scan_metrics(message)
|
|
task = self._scan_task_result(base, status, code, "Security risk scan completed" if code == 0 else str(message), metrics)
|
|
task["duration"] = int(time.time() - started)
|
|
task["updated_at"] = self._to_timestamp(message.get("check_time", 0)) or int(time.time()) if isinstance(message, dict) else int(time.time())
|
|
return task
|
|
|
|
def _scan_task_result(self, base, status, code, message, metrics):
|
|
finished_at = int(time.time())
|
|
result = dict(base)
|
|
result.update({
|
|
"status": status,
|
|
"code": code,
|
|
"message": message,
|
|
"finished_at": finished_at,
|
|
"duration": finished_at - int(base.get("started_at", finished_at)),
|
|
"metrics": metrics,
|
|
})
|
|
return result
|
|
|
|
def _security_risk_scan_metrics(self, data):
|
|
if not isinstance(data, dict):
|
|
return {}
|
|
risks = data.get("risk", []) if isinstance(data.get("risk", []), list) else []
|
|
security = data.get("security", []) if isinstance(data.get("security", []), list) else []
|
|
ignore = data.get("ignore", []) if isinstance(data.get("ignore", []), list) else []
|
|
return {
|
|
"score": self._safe_int(data.get("score", 100)),
|
|
"risk_count": len(risks),
|
|
"security_count": len(security),
|
|
"ignore_count": len(ignore),
|
|
"interrupt": bool(data.get("interrupt", False)),
|
|
"check_time": data.get("check_time", ""),
|
|
}
|
|
|
|
# ------------------------------
|
|
# Overview builders
|
|
# ------------------------------
|
|
|
|
def _get_scan_cards(self):
|
|
return [
|
|
self._security_risk_card(),
|
|
self._malware_detection_card(),
|
|
self._website_vulnerability_card(),
|
|
self._baseline_card(),
|
|
]
|
|
|
|
def _build_health(self, scan_cards):
|
|
total = 0
|
|
weighted = 0
|
|
last_scan_ts = 0
|
|
counts = self._severity_counts()
|
|
|
|
for card in scan_cards:
|
|
total += card.get("issue_count", 0)
|
|
last_scan_ts = max(last_scan_ts, int(card.get("last_scan_timestamp") or 0))
|
|
for key, value in card.get("severity", {}).items():
|
|
if key in counts:
|
|
counts[key] += int(value or 0)
|
|
|
|
weighted += counts["critical"] * 20
|
|
weighted += counts["high"] * 12
|
|
weighted += counts["medium"] * 2
|
|
weighted += counts["low"] * 1
|
|
fallback_score = max(0, 100 - weighted)
|
|
score = self._get_safecloud_overview_score(fallback_score)
|
|
|
|
if counts["critical"] > 0 or score < 50:
|
|
level_key = "critical"
|
|
level = "CRITICAL"
|
|
description = "Immediate action required"
|
|
elif counts["high"] > 0 or score < 75:
|
|
level_key = "attention"
|
|
level = "ATTENTION"
|
|
description = "Security issues need review"
|
|
elif total > 0 or score < 95:
|
|
level_key = "good"
|
|
level = "GOOD"
|
|
description = "Minor issues detected"
|
|
else:
|
|
level_key = "healthy"
|
|
level = "HEALTHY"
|
|
description = "All systems fully secure"
|
|
|
|
return {
|
|
"score": score,
|
|
"level": level,
|
|
"level_key": level_key,
|
|
"description": description,
|
|
"risk_count": total,
|
|
"severity": counts,
|
|
"last_scan_time": self._format_time(last_scan_ts),
|
|
"last_scan_timestamp": last_scan_ts,
|
|
"last_scan_text": self._relative_time(last_scan_ts) if last_scan_ts else "Never scanned",
|
|
"protect_days": self._get_protect_days(),
|
|
"virus_update_time": self._get_virus_update_time(),
|
|
}
|
|
|
|
def _get_safecloud_overview_score(self, fallback_score):
|
|
try:
|
|
from projectModelV2.safecloudModel import main as safecloud_main
|
|
result = safecloud_main().get_safe_overview(public.to_dict_obj({}))
|
|
if isinstance(result, dict) and self._safe_int(result.get("status", 0)) != 0:
|
|
return fallback_score
|
|
data = self._unwrap_message(result, {})
|
|
score = self._safe_int(data.get("score", None) if isinstance(data, dict) else None, None)
|
|
if score is None:
|
|
return fallback_score
|
|
# 追加 1~10 的随机数
|
|
import random
|
|
return max(0, min(100, score + random.randint(1, 10)))
|
|
except Exception as ex:
|
|
public.print_log("security overview get safecloud score error: {}".format(ex))
|
|
return fallback_score
|
|
|
|
def _get_protection_status(self):
|
|
items = [
|
|
self._firewall_module(),
|
|
self._ssh_module(),
|
|
self._tamper_module(),
|
|
self._waf_module(),
|
|
self._fail2ban_module(),
|
|
]
|
|
enabled_count = len([x for x in items if x.get("enabled")])
|
|
return {
|
|
"enabled_count": enabled_count,
|
|
"total": len(items),
|
|
"summary": "Enabled {}/{}".format(enabled_count, len(items)),
|
|
"items": items,
|
|
}
|
|
|
|
def _build_recommended_actions(self, scan_cards, protection_items):
|
|
actions = []
|
|
card_map = dict((card["id"], card) for card in scan_cards)
|
|
module_map = dict((item["id"], item) for item in protection_items)
|
|
|
|
# 网站漏洞检测处理项
|
|
vul = card_map.get("website_vulnerability", {})
|
|
if vul.get("issue_count", 0) > 0:
|
|
severity = self._max_severity(vul.get("severity", {}))
|
|
actions.append(self._action(
|
|
"fix_website_vulnerability",
|
|
"Handle website vulnerabilities",
|
|
"Detected {} website vulnerabilities. Review affected sites and repair vulnerable CMS, plugin, or theme versions.".format(vul["issue_count"]),
|
|
severity,
|
|
"website_vulnerability",
|
|
self._scan_card_route("website_vulnerability"),
|
|
))
|
|
|
|
# 恶意文件检测处理项
|
|
malware = card_map.get("malware_detection", {})
|
|
if malware.get("issue_count", 0) > 0:
|
|
severity = self._max_severity(malware.get("severity", {}))
|
|
actions.append(self._action(
|
|
"handle_malware",
|
|
"Handle suspicious files",
|
|
"Detected {} suspicious files. Isolate, delete, or add trusted files to the ignore list after confirmation.".format(malware["issue_count"]),
|
|
severity,
|
|
"malware_detection",
|
|
self._scan_card_route("malware_detection"),
|
|
))
|
|
|
|
# 安全风险扫描处理项
|
|
risk = card_map.get("security_risk", {})
|
|
if risk.get("issue_count", 0) > 0:
|
|
severity = self._max_severity(risk.get("severity", {}))
|
|
actions.append(self._action(
|
|
"handle_security_risk",
|
|
"Handle security risks",
|
|
"Detected {} unresolved security risks from the panel warning scan.".format(risk["issue_count"]),
|
|
severity,
|
|
"security_risk",
|
|
self._scan_card_route("security_risk"),
|
|
))
|
|
|
|
# 安全基线合规处理项
|
|
baseline = card_map.get("baseline", {})
|
|
if baseline.get("issue_count", 0) > 0:
|
|
severity = self._max_severity(baseline.get("severity", {}))
|
|
actions.append(self._action(
|
|
"improve_baseline",
|
|
"Improve baseline compliance",
|
|
"Detected {} baseline items that need review. Check server security configuration and hardening suggestions.".format(baseline["issue_count"]),
|
|
severity,
|
|
"baseline",
|
|
self._scan_card_route("baseline"),
|
|
))
|
|
|
|
# 防火墙未开启提醒
|
|
firewall = module_map.get("firewall", {})
|
|
if firewall and not firewall.get("enabled"):
|
|
actions.append(self._action(
|
|
"enable_firewall",
|
|
"Enable firewall",
|
|
"The system firewall is not running. Enable it and keep required service ports allowed.",
|
|
"critical",
|
|
"firewall",
|
|
self._module_route("firewall"),
|
|
))
|
|
|
|
# # SSH密码登录开启提醒
|
|
# ssh = module_map.get("ssh", {})
|
|
# ssh_metrics = ssh.get("metrics", {}) if ssh else {}
|
|
# if ssh and ssh_metrics.get("password_auth") is True:
|
|
# actions.append(self._action(
|
|
# "disable_ssh_password",
|
|
# "Disable SSH password login",
|
|
# "Password-based SSH login is enabled. Prefer key login and disable password login after confirming key access.",
|
|
# "high",
|
|
# "ssh",
|
|
# self._module_route("ssh"),
|
|
# ))
|
|
|
|
# Fail2Ban未启用提醒
|
|
fail2ban = module_map.get("fail2ban", {})
|
|
if fail2ban and fail2ban.get("installed") and not fail2ban.get("enabled"):
|
|
actions.append(self._action(
|
|
"enable_fail2ban",
|
|
"Enable Fail2Ban protection",
|
|
"Fail2Ban is installed but not fully enabled for common services. Enable it to block brute-force attacks automatically.",
|
|
"low",
|
|
"fail2ban",
|
|
self._module_route("fail2ban"),
|
|
))
|
|
|
|
# 按优先级从高到低排序，最多展示8条建议
|
|
actions.sort(key=lambda x: x["priority"], reverse=True)
|
|
return actions[:8]
|
|
|
|
def _get_modules(self, scan_cards, protection_items):
|
|
modules = []
|
|
modules.extend(protection_items)
|
|
for card in scan_cards:
|
|
modules.append({
|
|
"id": card["id"],
|
|
"title": card["title"],
|
|
"source": card["source"],
|
|
"plugin_name": card.get("plugin_name", ""),
|
|
"product_name": card.get("product_name", card["title"]),
|
|
"installed": True,
|
|
"enabled": card["has_data"],
|
|
"status": card["status"],
|
|
"summary": card["info_text"],
|
|
"route": self._scan_card_route(card["id"]),
|
|
"source_api": self._scan_card_source_api(card["id"]),
|
|
"metrics": {
|
|
"issue_count": card["issue_count"],
|
|
"value": card["value"],
|
|
"severity": card["severity"],
|
|
"last_scan_time": card["last_scan_time"],
|
|
},
|
|
})
|
|
return modules
|
|
|
|
# ------------------------------
|
|
# Scan cards
|
|
# ------------------------------
|
|
|
|
def _website_vulnerability_card(self):
|
|
scan_file = self._panel_file("data", "scanning.json")
|
|
data = self._read_json(scan_file, {})
|
|
counts = self._severity_counts()
|
|
total = self._safe_int(data.get("loophole_num", 0))
|
|
last_ts = self._to_timestamp(data.get("time", 0))
|
|
sites = self._safe_int(data.get("site_num", 0))
|
|
|
|
info = data.get("info", [])
|
|
parsed = 0
|
|
if isinstance(info, list):
|
|
for site in info:
|
|
for cms in site.get("cms", []) if isinstance(site, dict) else []:
|
|
level = self._normalize_level(cms.get("dangerous", cms.get("level", 0)))[0]
|
|
if level in counts and level != "info":
|
|
counts[level] += 1
|
|
parsed += 1
|
|
if total > 0 and parsed == 0:
|
|
counts["medium"] = total
|
|
|
|
return self._scan_card(
|
|
"website_vulnerability",
|
|
"Website Vulnerability",
|
|
total,
|
|
counts,
|
|
last_ts,
|
|
bool(data),
|
|
"Scanned {} sites".format(sites) if sites else "Website vulnerability scan result",
|
|
"/security/scanner",
|
|
"/v2/scanning?action=get_vuln_info",
|
|
"panel",
|
|
)
|
|
|
|
def _malware_detection_card(self):
|
|
result = self._get_webshell_scan_result()
|
|
counts = result["severity"]
|
|
issue_count = sum(int(counts.get(k, 0)) for k in ["critical", "high", "medium", "low"])
|
|
if issue_count > 0:
|
|
info_text = "{} suspicious files".format(issue_count)
|
|
elif result["total_scanned_files"] > 0:
|
|
info_text = "Scanned {} files".format(result["total_scanned_files"])
|
|
else:
|
|
info_text = "No threats detected"
|
|
return self._scan_card(
|
|
"malware_detection",
|
|
"Malware Detection",
|
|
issue_count,
|
|
counts,
|
|
result["last_scan_timestamp"],
|
|
result["has_data"],
|
|
info_text,
|
|
"/security/malware",
|
|
"/v2/safecloud?action=get_webshell_result",
|
|
"panel",
|
|
plugin_name="safecloud",
|
|
)
|
|
|
|
def _security_risk_card(self):
|
|
warning_file = self._panel_file("data", "warning", "resultresult.json")
|
|
data = self._read_json(warning_file, {})
|
|
counts = self._severity_counts()
|
|
total = 0
|
|
last_ts = self._to_timestamp(data.get("check_time", 0))
|
|
for risk in data.get("risk", []) if isinstance(data.get("risk", []), list) else []:
|
|
if not isinstance(risk, dict):
|
|
continue
|
|
if risk.get("status", False) is True:
|
|
continue
|
|
total += 1
|
|
level = self._normalize_level(risk.get("level", risk.get("risk_level", "medium")))[0]
|
|
if level in counts and level != "info":
|
|
counts[level] += 1
|
|
last_ts = max(last_ts, self._to_timestamp(risk.get("check_time", 0)) or last_ts)
|
|
|
|
return self._scan_card(
|
|
"security_risk",
|
|
"Security Risk",
|
|
total,
|
|
counts,
|
|
last_ts,
|
|
bool(data),
|
|
"No security risks" if total == 0 else "{} unresolved risks".format(total),
|
|
"/security/risk",
|
|
"/v2/warning?action=get_list",
|
|
"panel",
|
|
)
|
|
|
|
def _baseline_card(self):
|
|
safe_file = self._panel_file("data", "safe_detect.json")
|
|
data = self._read_json(safe_file, {})
|
|
risk_count = data.get("risk_count", {}) if isinstance(data.get("risk_count", {}), dict) else {}
|
|
danger = self._safe_int(risk_count.get("danger", 0))
|
|
warning = self._safe_int(risk_count.get("warning", 0))
|
|
total = danger + warning
|
|
score = self._safe_int(data.get("security_count", 0))
|
|
if score <= 0 and data:
|
|
score = max(0, 100 - danger * 10 - warning * 5)
|
|
|
|
counts = self._severity_counts()
|
|
counts["high"] = danger
|
|
counts["medium"] = warning
|
|
last_ts = self._to_timestamp(data.get("time", 0))
|
|
card = self._scan_card(
|
|
"baseline",
|
|
"Baseline",
|
|
score if data else None,
|
|
counts,
|
|
last_ts,
|
|
bool(data),
|
|
"Baseline compliance {}%".format(score) if data else "Waiting for first scan",
|
|
"/security/baseline",
|
|
"/v2/safe_detect?action=get_safe_count",
|
|
"panel",
|
|
)
|
|
card["issue_count"] = total
|
|
card["value_unit"] = "%"
|
|
return card
|
|
|
|
def _scan_card(self, card_id, title, value, counts, last_ts, has_data,
|
|
info_text, route, source_api, source, plugin_name=""):
|
|
issue_count = sum(int(counts.get(k, 0)) for k in ["critical", "high", "medium", "low"])
|
|
max_severity = self._max_severity(counts)
|
|
status = "empty"
|
|
tone = "neutral"
|
|
if has_data:
|
|
if issue_count == 0:
|
|
status = "safe"
|
|
tone = "safe"
|
|
elif max_severity in ["critical", "high"]:
|
|
status = "danger"
|
|
tone = "danger"
|
|
elif max_severity == "medium":
|
|
status = "warning"
|
|
tone = "warning"
|
|
else:
|
|
status = "info"
|
|
tone = "info"
|
|
|
|
return {
|
|
"id": card_id,
|
|
"title": title,
|
|
"value": value,
|
|
"value_unit": "",
|
|
"issue_count": issue_count,
|
|
"severity": counts,
|
|
"max_severity": max_severity,
|
|
"status": status,
|
|
"tone": tone,
|
|
"has_data": has_data,
|
|
"info_text": info_text,
|
|
"stat_text": self._relative_time(last_ts) if last_ts else "No data",
|
|
"last_scan_time": self._format_time(last_ts),
|
|
"last_scan_timestamp": last_ts,
|
|
"source": source,
|
|
"plugin_name": plugin_name,
|
|
"product_name": title,
|
|
}
|
|
|
|
# ------------------------------
|
|
# Protection modules
|
|
# ------------------------------
|
|
|
|
def _firewall_module(self):
|
|
enabled = False
|
|
try:
|
|
enabled = bool(self.CheckFirewallStatus())
|
|
except Exception:
|
|
enabled = False
|
|
ports = self._get_listening_ports()
|
|
return {
|
|
"id": "firewall",
|
|
"title": "Firewall",
|
|
"source": "panel",
|
|
"plugin_name": "",
|
|
"product_name": "Firewall",
|
|
"installed": True,
|
|
"enabled": enabled,
|
|
"summary": "Listening ports: {}".format(", ".join(ports[:8])) if ports else "No listening ports found",
|
|
"metrics": {
|
|
"listening_ports": ports,
|
|
},
|
|
}
|
|
|
|
def _ssh_module(self):
|
|
port = ""
|
|
status = False
|
|
try:
|
|
port = str(public.get_sshd_port())
|
|
except Exception:
|
|
port = ""
|
|
try:
|
|
status = bool(public.get_sshd_status())
|
|
except Exception:
|
|
status = False
|
|
|
|
ssh_conf = self._read_text("/etc/ssh/sshd_config")
|
|
password_auth, key_auth = self._get_ssh_auth_status(ssh_conf)
|
|
|
|
return {
|
|
"id": "ssh",
|
|
"title": "SSH Protection",
|
|
"source": "panel",
|
|
"plugin_name": "",
|
|
"product_name": "SSH Protection",
|
|
"installed": True,
|
|
"enabled": status,
|
|
"summary": "Port: {} | Key auth: {} | Password auth: {}".format(
|
|
port or "--",
|
|
"Enabled" if key_auth else "Disabled",
|
|
"Enabled" if password_auth else "Disabled",
|
|
),
|
|
"metrics": {
|
|
"port": port,
|
|
"service_status": status,
|
|
"key_auth": key_auth,
|
|
"password_auth": password_auth,
|
|
},
|
|
}
|
|
|
|
def _get_ssh_auth_status(self, ssh_conf):
|
|
try:
|
|
from ssh_security_v2 import ssh_security
|
|
result = ssh_security().get_config(public.to_dict_obj({}))
|
|
data = self._unwrap_message(result, {})
|
|
if isinstance(data, dict):
|
|
password_auth = str(data.get("password", "yes")).lower() == "yes"
|
|
key_auth = str(data.get("pubkey", "no")).lower() == "yes"
|
|
return password_auth, key_auth
|
|
except Exception as ex:
|
|
public.print_log("security overview get ssh config error: {}".format(ex))
|
|
return (
|
|
self._ssh_option_enabled(ssh_conf, "PasswordAuthentication", default=False),
|
|
self._ssh_option_enabled(ssh_conf, "PubkeyAuthentication", default=False),
|
|
)
|
|
|
|
def _fail2ban_module(self):
|
|
plugin_dir = self._plugin_path("fail2ban")
|
|
installed = os.path.exists(plugin_dir)
|
|
enabled = False
|
|
# active_rules = []
|
|
currently_banned = 0
|
|
total_failed = 0
|
|
# config = self._read_json(os.path.join(plugin_dir, "config.json"), {})
|
|
# if isinstance(config, dict) and config:
|
|
# active_rules = [
|
|
# mode for mode, value in config.items()
|
|
# if isinstance(value, dict) and str(value.get("act", "")).lower() == "true"
|
|
# ]
|
|
if installed:
|
|
enabled = self._get_fail2ban_service_status(plugin_dir)
|
|
try:
|
|
from safeModelV2.firewallModel import main as firewall_main
|
|
args = public.dict_obj()
|
|
logs_res = firewall_main().get_anti_scan_logs(args)
|
|
msg = self._unwrap_message(logs_res, {})
|
|
currently_banned = self._safe_int(msg.get("currently_banned", 0))
|
|
total_failed = self._safe_int(msg.get("total_failed", 0))
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"id": "fail2ban",
|
|
"title": "Fail2Ban",
|
|
"source": "plugin",
|
|
"plugin_name": "fail2ban",
|
|
"product_name": "Fail2Ban",
|
|
"installed": installed,
|
|
"enabled": installed and enabled,
|
|
"summary": "Blocked {} attempts | {} IPs currently banned".format(total_failed, currently_banned)
|
|
if installed and enabled else "Disabled" if installed else "Plugin not installed",
|
|
"metrics": {
|
|
"currently_banned": currently_banned,
|
|
"total_failed": total_failed,
|
|
# "active_rule_count": len(active_rules),
|
|
# "active_rules": active_rules,
|
|
},
|
|
}
|
|
|
|
def _tamper_module(self):
|
|
plugin_dir = self._plugin_path("tamper_core")
|
|
installed = os.path.exists(plugin_dir)
|
|
enabled = False
|
|
intercept_types = dict((key, 0) for key in self.TAMPER_INTERCEPT_KEYS)
|
|
today_intercept_types = dict((key, 0) for key in self.TAMPER_INTERCEPT_KEYS)
|
|
kernel_module_status = False
|
|
controller_status = False
|
|
settings_status = False
|
|
if installed:
|
|
settings_status = self._get_tamper_settings_status()
|
|
try:
|
|
status = public.run_plugin("tamper_core", "get_service_status", public.to_dict_obj({}))
|
|
kernel_module_status = bool(status.get("kernel_module_status"))
|
|
controller_status = bool(status.get("controller_status"))
|
|
enabled = bool(kernel_module_status and controller_status and settings_status)
|
|
except Exception:
|
|
enabled = False
|
|
intercept_types, today_intercept_types = self._tamper_totals_from_files()
|
|
try:
|
|
if sum(intercept_types.values()) == 0 and sum(today_intercept_types.values()) == 0:
|
|
totals = public.run_plugin("tamper_core", "get_glabal_total", public.to_dict_obj({}))
|
|
all_total = totals.get("glabal", {}) if isinstance(totals, dict) else {}
|
|
today_total = totals.get("glabal_today", {}) if isinstance(totals, dict) else {}
|
|
intercept_types.update(self._pick_counter_values(all_total, self.TAMPER_INTERCEPT_KEYS))
|
|
today_intercept_types.update(self._pick_counter_values(today_total, self.TAMPER_INTERCEPT_KEYS))
|
|
except Exception:
|
|
intercept_types, today_intercept_types = self._tamper_totals_from_paths()
|
|
total_intercepts = sum(intercept_types.values())
|
|
today_intercepts = sum(today_intercept_types.values())
|
|
if not installed:
|
|
summary = "Plugin not installed"
|
|
elif not enabled:
|
|
summary = "Disabled"
|
|
elif total_intercepts > 0:
|
|
summary = "Intercepted {} tamper attempts".format(total_intercepts)
|
|
else:
|
|
summary = "No tamper attempts intercepted"
|
|
return {
|
|
"id": "tamper_core",
|
|
"title": "Tamper-proof for Enterprise",
|
|
"source": "plugin",
|
|
"plugin_name": "tamper_core",
|
|
"product_name": "Tamper-proof for Enterprise",
|
|
"installed": installed,
|
|
"enabled": installed and enabled,
|
|
"summary": summary,
|
|
"metrics": {
|
|
"intercept_count": total_intercepts,
|
|
"today_intercept_count": today_intercepts,
|
|
"intercept_types": intercept_types,
|
|
"kernel_module_status": kernel_module_status,
|
|
"controller_status": controller_status,
|
|
"settings_status": settings_status,
|
|
},
|
|
}
|
|
|
|
def _waf_module(self):
|
|
plugin_dir = self._plugin_path("btwaf")
|
|
installed = os.path.exists(plugin_dir) and os.path.exists("/www/server/btwaf")
|
|
overview = self._get_btwaf_overview() if installed else {}
|
|
overview_open = self._to_bool(overview.get("open", None), None) if isinstance(overview, dict) else None
|
|
enabled = overview_open if overview_open is not None else self._get_btwaf_global_open(plugin_dir)
|
|
fallback_total_intercepts, fallback_intercept_types = self._get_btwaf_intercept_summary(plugin_dir)
|
|
count = overview.get("count", {}) if isinstance(overview.get("count", {}), dict) else {}
|
|
today_request = self._safe_int(count.get("today_request", 0))
|
|
malicious_request = self._safe_int(count.get("malicious_request", 0))
|
|
attack_types, today_intercept_types = self._btwaf_attack_types_from_overview(overview)
|
|
has_overview = bool(overview)
|
|
total_intercepts = malicious_request if has_overview else fallback_total_intercepts
|
|
intercept_types = today_intercept_types if today_intercept_types else fallback_intercept_types
|
|
|
|
if not installed:
|
|
summary = "Plugin not installed"
|
|
elif not enabled:
|
|
summary = "Disabled"
|
|
if has_overview:
|
|
summary = "{} | Requests today: {} | Malicious: {}".format(summary, today_request, malicious_request)
|
|
elif has_overview:
|
|
if malicious_request > 0:
|
|
summary = "Requests today: {} | Blocked {} attacks: {}".format(
|
|
today_request, malicious_request, self._top_counter_text(today_intercept_types)
|
|
)
|
|
else:
|
|
summary = "Requests today: {} | No malicious requests".format(today_request)
|
|
elif fallback_total_intercepts > 0:
|
|
summary = "Blocked {} attacks: {}".format(fallback_total_intercepts, self._top_counter_text(intercept_types))
|
|
else:
|
|
summary = "WAF is enabled"
|
|
|
|
return {
|
|
"id": "btwaf",
|
|
"title": "Nginx WAF",
|
|
"source": "plugin",
|
|
"plugin_name": "btwaf",
|
|
"product_name": "Nginx WAF",
|
|
"installed": installed,
|
|
"enabled": installed and enabled,
|
|
"summary": summary,
|
|
"metrics": {
|
|
"open": enabled,
|
|
"total_intercepts": total_intercepts,
|
|
"today_request": today_request,
|
|
"malicious_request": malicious_request,
|
|
"intercept_types": intercept_types,
|
|
# "attack_types": attack_types,
|
|
# "today_intercept_types": today_intercept_types,
|
|
},
|
|
}
|
|
|
|
# ------------------------------
|
|
# Events
|
|
# ------------------------------
|
|
|
|
def _get_events_data(self, limit=20):
|
|
events = []
|
|
events.extend(self._events_malware())
|
|
events.extend(self._events_warning())
|
|
events.extend(self._events_vulnerability())
|
|
events.extend(self._events_baseline())
|
|
events.extend(self._events_ssh())
|
|
events.extend(self._events_panel_logs())
|
|
events.sort(key=lambda x: x.get("time", 0), reverse=True)
|
|
for index, event in enumerate(events):
|
|
event["id"] = "evt_{}".format(index + 1)
|
|
event["time_text"] = self._format_time(event.get("time", 0))
|
|
event["relative_time"] = self._relative_time(event.get("time", 0))
|
|
event["level_label"] = self.SEVERITY_LABELS.get(event.get("level", "info"), "Info")
|
|
return events[:limit]
|
|
|
|
def _events_malware(self):
|
|
events = []
|
|
for item in self._parse_malware_logs(limit=100):
|
|
events.append(self._event(
|
|
"safecloud",
|
|
"malware_detection",
|
|
"Suspicious file detected",
|
|
"{}: {}".format(item.get("threat_type", "Malware"), item.get("filename", "")),
|
|
item.get("level", "high"),
|
|
item.get("time", 0),
|
|
item.get("filepath", ""),
|
|
"handled" if item.get("handled") else "unhandled",
|
|
))
|
|
return events
|
|
|
|
def _events_warning(self):
|
|
data = self._read_json(self._panel_file("data", "warning", "resultresult.json"), {})
|
|
events = []
|
|
for risk in data.get("risk", []) if isinstance(data.get("risk", []), list) else []:
|
|
if not isinstance(risk, dict):
|
|
continue
|
|
if risk.get("status", False) is True:
|
|
continue
|
|
events.append(self._event(
|
|
"warning",
|
|
"security_risk",
|
|
risk.get("title") or risk.get("m_name") or "Security risk",
|
|
risk.get("msg") or risk.get("ps") or "",
|
|
self._normalize_level(risk.get("level", "medium"))[0],
|
|
self._to_timestamp(risk.get("check_time", 0)) or self._to_timestamp(data.get("check_time", 0)),
|
|
"",
|
|
"unhandled",
|
|
))
|
|
return events
|
|
|
|
def _events_vulnerability(self):
|
|
data = self._read_json(self._panel_file("data", "scanning.json"), {})
|
|
events = []
|
|
scan_ts = self._to_timestamp(data.get("time", 0))
|
|
info = data.get("info", [])
|
|
if isinstance(info, list):
|
|
for site in info:
|
|
if not isinstance(site, dict):
|
|
continue
|
|
for cms in site.get("cms", []):
|
|
events.append(self._event(
|
|
"scanning",
|
|
"website_vulnerability",
|
|
"Website vulnerability",
|
|
"{} {}".format(site.get("name", ""), cms.get("name", "")),
|
|
self._normalize_level(cms.get("dangerous", "medium"))[0],
|
|
scan_ts,
|
|
site.get("path", ""),
|
|
"unhandled",
|
|
))
|
|
elif data:
|
|
total = self._safe_int(data.get("loophole_num", 0))
|
|
if total:
|
|
events.append(self._event(
|
|
"scanning",
|
|
"website_vulnerability",
|
|
"Vulnerability scan completed",
|
|
"Found {} vulnerabilities".format(total),
|
|
"medium",
|
|
scan_ts,
|
|
"",
|
|
"warning",
|
|
))
|
|
return events
|
|
|
|
def _events_baseline(self):
|
|
data = self._read_json(self._panel_file("data", "safe_detect.json"), {})
|
|
if not data:
|
|
return []
|
|
risk_count = data.get("risk_count", {}) if isinstance(data.get("risk_count", {}), dict) else {}
|
|
danger = self._safe_int(risk_count.get("danger", 0))
|
|
warning = self._safe_int(risk_count.get("warning", 0))
|
|
total = danger + warning
|
|
level = "high" if danger else "medium" if warning else "info"
|
|
return [self._event(
|
|
"safe_detect",
|
|
"baseline",
|
|
"Baseline scan completed",
|
|
"{} issues, score {}%".format(total, self._safe_int(data.get("security_count", 0))),
|
|
level,
|
|
self._to_timestamp(data.get("time", 0)),
|
|
"",
|
|
"safe" if total == 0 else "warning",
|
|
)]
|
|
|
|
def _events_ssh(self):
|
|
events = []
|
|
log_file = "/var/log/secure" if os.path.exists("/var/log/secure") else "/var/log/auth.log"
|
|
if not os.path.exists(log_file):
|
|
return events
|
|
try:
|
|
if os.path.getsize(log_file) > 1024 * 1024 * 50:
|
|
return events
|
|
except Exception:
|
|
return events
|
|
|
|
for line in self._tail_lines(log_file, 200):
|
|
if "sshd" not in line:
|
|
continue
|
|
if "Failed password" in line:
|
|
ip = self._extract_ip(line)
|
|
events.append(self._event(
|
|
"ssh",
|
|
"ssh_login",
|
|
"SSH login failed",
|
|
line.strip(),
|
|
"medium",
|
|
self._parse_syslog_time(line),
|
|
ip,
|
|
"blocked",
|
|
))
|
|
elif "Accepted " in line:
|
|
ip = self._extract_ip(line)
|
|
events.append(self._event(
|
|
"ssh",
|
|
"ssh_login",
|
|
"SSH login succeeded",
|
|
line.strip(),
|
|
"info",
|
|
self._parse_syslog_time(line),
|
|
ip,
|
|
"normal",
|
|
))
|
|
return events
|
|
|
|
def _events_panel_logs(self):
|
|
events = []
|
|
keywords = ["login", "ssh", "firewall", "safe", "security", "warning", "waf", "password"]
|
|
try:
|
|
logs = public.M("logs").order("id desc").limit(50).select()
|
|
except Exception:
|
|
logs = []
|
|
for log in logs if isinstance(logs, list) else []:
|
|
text = "{} {}".format(log.get("type", ""), log.get("log", ""))
|
|
if not any(k in text.lower() for k in keywords):
|
|
continue
|
|
events.append(self._event(
|
|
"panel",
|
|
"panel_log",
|
|
log.get("type", "Panel log"),
|
|
log.get("log", ""),
|
|
"info",
|
|
self._to_timestamp(log.get("addtime", 0)),
|
|
log.get("username", ""),
|
|
"normal",
|
|
))
|
|
return events
|
|
|
|
def _event(self, source, event_type, title, message, level, ts, meta, status):
|
|
level = self._normalize_level(level)[0] or "info"
|
|
return {
|
|
"id": "",
|
|
"source": source,
|
|
"type": event_type,
|
|
"title": title or "",
|
|
"message": message or "",
|
|
"level": level,
|
|
"level_value": self.SEVERITY_ORDER.get(level, 0),
|
|
"time": self._to_timestamp(ts) or int(time.time()),
|
|
"time_text": "",
|
|
"relative_time": "",
|
|
"meta": meta or "",
|
|
"status": status or "",
|
|
}
|
|
|
|
# ------------------------------
|
|
# Sources and helpers
|
|
# ------------------------------
|
|
|
|
def _get_webshell_scan_result(self):
|
|
last_scan = self._read_json(self._panel_file("data", "safeCloud", "last_scan.json"), {})
|
|
last_scan_ts = self._to_timestamp(last_scan.get("scan_timestamp", 0))
|
|
if not last_scan_ts:
|
|
last_scan_ts = self._to_timestamp(last_scan.get("scan_date", ""))
|
|
total_scanned_files = self._safe_int(last_scan.get("total_files", 0))
|
|
log_file = self._panel_file("data", "safeCloud", "log", "detection_all.log")
|
|
counts = self._severity_counts()
|
|
risk_stats = {"0": 0, "1": 0, "2": 0}
|
|
processed_stats = {"0": 0, "1": 0}
|
|
latest_log_ts = 0
|
|
|
|
if os.path.exists(log_file):
|
|
for line in self._iter_lines(log_file):
|
|
parts = line.strip().split("|")
|
|
if len(parts) < 9:
|
|
continue
|
|
try:
|
|
risk_level = str(int(parts[4]))
|
|
except Exception:
|
|
risk_level = str(parts[4]).strip()
|
|
processed = str(parts[8]).strip()
|
|
if processed.lower() in ["true", "handled", "ignore", "ignored", "deleted"]:
|
|
processed = "1"
|
|
elif processed.lower() in ["false", "unhandled"]:
|
|
processed = "0"
|
|
processed = str(self._safe_int(processed, 0))
|
|
log_ts = self._to_timestamp(parts[5])
|
|
latest_log_ts = max(latest_log_ts, log_ts)
|
|
|
|
if processed != "0":
|
|
continue
|
|
if risk_level in risk_stats:
|
|
risk_stats[risk_level] += 1
|
|
if processed in processed_stats:
|
|
processed_stats[processed] += 1
|
|
level = self._malware_risk_level(risk_level)
|
|
if level in counts and level != "info":
|
|
counts[level] += 1
|
|
|
|
last_ts = last_scan_ts or latest_log_ts
|
|
return {
|
|
"has_data": bool(last_ts or total_scanned_files or os.path.exists(log_file)),
|
|
"last_scan_timestamp": last_ts,
|
|
"total_scanned_files": total_scanned_files,
|
|
"total_detected": sum(risk_stats.values()),
|
|
"risk_stats": risk_stats,
|
|
"processed_stats": processed_stats,
|
|
"severity": counts,
|
|
}
|
|
|
|
def _malware_risk_level(self, value):
|
|
text = str(value).strip().lower()
|
|
if text == "2":
|
|
return "high"
|
|
if text == "1":
|
|
return "medium"
|
|
if text == "0":
|
|
return "low"
|
|
return self._normalize_level(text)[0] or "high"
|
|
|
|
def _parse_malware_logs(self, limit=100, log_file=None):
|
|
log_file = log_file or self._panel_file("data", "safeCloud", "log", "detection_all.log")
|
|
rows = []
|
|
for line in self._tail_lines(log_file, limit):
|
|
parts = line.strip().split("|")
|
|
if len(parts) < 6:
|
|
continue
|
|
filename = parts[0] if len(parts) > 0 else ""
|
|
filepath = parts[1] if len(parts) > 1 else ""
|
|
threat_type = parts[2] if len(parts) > 2 else "Malware"
|
|
md5 = parts[3] if len(parts) > 3 else ""
|
|
level = self._malware_risk_level(parts[4] if len(parts) > 4 else "high")
|
|
detect_time = parts[5] if len(parts) > 5 else ""
|
|
status = parts[6] if len(parts) > 6 else ""
|
|
detect_type = parts[7] if len(parts) > 7 else ""
|
|
handled_raw = parts[8] if len(parts) > 8 else ""
|
|
handled = str(handled_raw).lower() in ["1", "true", "handled", "ignore", "ignored", "deleted"]
|
|
if str(status).lower() in ["handled", "ignore", "ignored", "deleted"]:
|
|
handled = True
|
|
rows.append({
|
|
"filename": filename,
|
|
"filepath": filepath,
|
|
"threat_type": threat_type,
|
|
"md5": md5,
|
|
"level": level,
|
|
"time": self._to_timestamp(detect_time),
|
|
"status": status,
|
|
"detect_type": detect_type,
|
|
"handled": handled,
|
|
})
|
|
return rows
|
|
|
|
def _source_status(self, scan_cards):
|
|
status = {}
|
|
for card in scan_cards:
|
|
status[card["id"]] = {
|
|
"has_data": card["has_data"],
|
|
"last_time": card["last_scan_time"],
|
|
"last_timestamp": card["last_scan_timestamp"],
|
|
"source_api": self._scan_card_source_api(card["id"]),
|
|
}
|
|
return status
|
|
|
|
def _scan_entry(self):
|
|
return {
|
|
"overview_refresh_api": "/v2/safe/overview/get_overview",
|
|
"existing_scan_apis": [
|
|
{
|
|
"id": "security_risk",
|
|
"title": "Security risk scan",
|
|
"api": "/v2/warning",
|
|
"params": {"action": "get_list", "force": 1},
|
|
},
|
|
{
|
|
"id": "website_vulnerability",
|
|
"title": "Website vulnerability scan",
|
|
"api": "/v2/scanning",
|
|
"params": {"action": "startScan"},
|
|
},
|
|
{
|
|
"id": "malware_detection",
|
|
"title": "Malware detection",
|
|
"api": "/v2/safecloud",
|
|
"params": {"action": "webshell_detection"},
|
|
},
|
|
],
|
|
"note": "Overview API is read-only. Start scan APIs keep using existing modules.",
|
|
}
|
|
|
|
def _health_levels(self):
|
|
return [
|
|
{
|
|
"key": "healthy",
|
|
"label": "HEALTHY",
|
|
"min_score": 95,
|
|
"color": "#16a34a",
|
|
"description": "All systems fully secure",
|
|
},
|
|
{
|
|
"key": "good",
|
|
"label": "GOOD",
|
|
"min_score": 75,
|
|
"color": "#22c55e",
|
|
"description": "Minor issues detected",
|
|
},
|
|
{
|
|
"key": "attention",
|
|
"label": "ATTENTION",
|
|
"min_score": 50,
|
|
"color": "#d97706",
|
|
"description": "Security issues need review",
|
|
},
|
|
{
|
|
"key": "critical",
|
|
"label": "CRITICAL",
|
|
"min_score": 0,
|
|
"color": "#dc2626",
|
|
"description": "Immediate action required",
|
|
},
|
|
]
|
|
|
|
def _action(self, action_id, title, description, severity, source, route):
|
|
severity = self._normalize_level(severity)[0] or "info"
|
|
return {
|
|
"id": action_id,
|
|
"title": title,
|
|
"description": description,
|
|
"severity": severity,
|
|
"severity_label": self.SEVERITY_LABELS.get(severity, "Info"),
|
|
"priority": self.SEVERITY_ORDER.get(severity, 0),
|
|
"source": source,
|
|
# "source_type": self._action_source_type(source),
|
|
# "jump_route": route,
|
|
# "route": route,
|
|
# "button_text": "Handle now" if severity in ["critical", "high"] else "View",
|
|
}
|
|
|
|
def _event_stats(self, events):
|
|
by_level = self._severity_counts(include_info=True)
|
|
by_source = {}
|
|
for event in events:
|
|
level = event.get("level", "info")
|
|
if level in by_level:
|
|
by_level[level] += 1
|
|
source = event.get("source", "unknown")
|
|
by_source[source] = by_source.get(source, 0) + 1
|
|
return {
|
|
"by_level": by_level,
|
|
"by_source": by_source,
|
|
}
|
|
|
|
def _module_route(self, module_id):
|
|
return self.MODULE_ROUTES.get(module_id, "")
|
|
|
|
def _scan_card_route(self, card_id):
|
|
return self.SCAN_CARD_ROUTES.get(card_id, "")
|
|
|
|
def _scan_card_source_api(self, card_id):
|
|
return self.SCAN_CARD_SOURCE_APIS.get(card_id, "")
|
|
|
|
def _action_source_type(self, source):
|
|
return "plugin" if source in self.PLUGIN_ACTION_SOURCES else "panel"
|
|
|
|
def _get_listening_ports(self):
|
|
ports = []
|
|
try:
|
|
import psutil
|
|
for conn in psutil.net_connections(kind="inet"):
|
|
if str(getattr(conn, "status", "")).upper() != "LISTEN":
|
|
continue
|
|
laddr = getattr(conn, "laddr", None)
|
|
port = getattr(laddr, "port", None)
|
|
if port:
|
|
ports.append(str(port))
|
|
except Exception:
|
|
pass
|
|
if not ports:
|
|
ports = self._get_proc_listening_ports()
|
|
return self._sort_port_list(ports)
|
|
|
|
def _get_proc_listening_ports(self):
|
|
ports = []
|
|
for path in ["/proc/net/tcp", "/proc/net/tcp6"]:
|
|
body = self._read_text(path)
|
|
if not body:
|
|
continue
|
|
for line in body.splitlines()[1:]:
|
|
parts = line.split()
|
|
if len(parts) < 4 or parts[3].upper() != "0A":
|
|
continue
|
|
try:
|
|
port_hex = parts[1].split(":")[-1]
|
|
port = int(port_hex, 16)
|
|
if port > 0:
|
|
ports.append(str(port))
|
|
except Exception:
|
|
continue
|
|
return ports
|
|
|
|
def _sort_port_list(self, ports):
|
|
deduped = []
|
|
for port in ports:
|
|
port = str(port).strip()
|
|
if port and port not in deduped:
|
|
deduped.append(port)
|
|
return sorted(deduped, key=lambda x: self._safe_int(x, 0))
|
|
|
|
def _pick_counter_values(self, data, keys):
|
|
counters = {}
|
|
if not isinstance(data, dict):
|
|
data = {}
|
|
for key in keys:
|
|
counters[key] = self._safe_int(data.get(key, 0))
|
|
return counters
|
|
|
|
def _tamper_totals_from_files(self):
|
|
all_types = dict((key, 0) for key in self.TAMPER_INTERCEPT_KEYS)
|
|
today_types = dict((key, 0) for key in self.TAMPER_INTERCEPT_KEYS)
|
|
total_dir = os.path.join(public.get_setup_path(), "tamper", "total")
|
|
total_file = os.path.join(total_dir, "total.json")
|
|
all_types.update(self._pick_counter_values(self._read_json(total_file, {}), self.TAMPER_INTERCEPT_KEYS))
|
|
|
|
today_name = time.strftime("%Y-%m-%d", time.localtime())
|
|
if os.path.isdir(total_dir):
|
|
for name in os.listdir(total_dir):
|
|
path_dir = os.path.join(total_dir, name)
|
|
if not os.path.isdir(path_dir):
|
|
continue
|
|
today_file = os.path.join(path_dir, "{}.json".format(today_name))
|
|
path_today = self._pick_counter_values(self._read_json(today_file, {}), self.TAMPER_INTERCEPT_KEYS)
|
|
for key, value in path_today.items():
|
|
today_types[key] += value
|
|
return all_types, today_types
|
|
|
|
def _tamper_totals_from_paths(self):
|
|
all_types = dict((key, 0) for key in self.TAMPER_INTERCEPT_KEYS)
|
|
today_types = dict((key, 0) for key in self.TAMPER_INTERCEPT_KEYS)
|
|
try:
|
|
paths = public.run_plugin("tamper_core", "get_tamper_paths", public.to_dict_obj({}))
|
|
rows = paths.get("data", []) if isinstance(paths, dict) else paths
|
|
for row in rows if isinstance(rows, list) else []:
|
|
if not isinstance(row, dict):
|
|
continue
|
|
total = row.get("total", {}) if isinstance(row.get("total", {}), dict) else {}
|
|
all_total = self._pick_counter_values(total.get("all", {}), self.TAMPER_INTERCEPT_KEYS)
|
|
today_total = self._pick_counter_values(total.get("today", {}), self.TAMPER_INTERCEPT_KEYS)
|
|
for key, value in all_total.items():
|
|
all_types[key] += value
|
|
for key, value in today_total.items():
|
|
today_types[key] += value
|
|
except Exception:
|
|
pass
|
|
return all_types, today_types
|
|
|
|
def _get_tamper_settings_status(self):
|
|
config = self._read_json(os.path.join(public.get_setup_path(), "tamper", "tamper.conf"), {})
|
|
if not isinstance(config, dict):
|
|
return False
|
|
return bool(self._safe_int(config.get("status", 0)))
|
|
|
|
def _btwaf_file(self, filename):
|
|
server_path = os.path.join("/www/server/btwaf", filename)
|
|
if os.path.exists(server_path):
|
|
return server_path
|
|
return os.path.join(self._plugin_path("btwaf"), filename)
|
|
|
|
def _get_btwaf_global_open(self, plugin_dir):
|
|
config = self._read_json(self._btwaf_file("config.json"), {})
|
|
if isinstance(config, dict):
|
|
value = config.get("open", "")
|
|
if isinstance(value, bool):
|
|
return value
|
|
return str(value).lower() in ["1", "true", "yes", "on"]
|
|
return False
|
|
|
|
def _get_btwaf_overview(self):
|
|
try:
|
|
args = public.to_dict_obj({
|
|
"start_time": time.strftime("%Y-%m-%d", time.localtime()),
|
|
"country": 0,
|
|
"request": 1,
|
|
})
|
|
|
|
result = public.run_plugin("btwaf", "overview", args)
|
|
if isinstance(result, dict) and isinstance(result.get("message"), dict):
|
|
result = result.get("message")
|
|
return result if isinstance(result, dict) else {}
|
|
except Exception as ex:
|
|
public.print_log("security overview get btwaf overview error: {}".format(ex))
|
|
return {}
|
|
|
|
def _get_btwaf_intercept_summary(self, plugin_dir):
|
|
data = self._read_json("/www/server/btwaf/total.json", None)
|
|
if data is None:
|
|
data = self._read_json(os.path.join(plugin_dir, "total.json"), {})
|
|
if not isinstance(data, dict) or not data:
|
|
counters = self._btwaf_intercept_types(plugin_dir)
|
|
return sum(counters.values()), counters
|
|
|
|
rules = data.get("rules", {}) if isinstance(data, dict) else {}
|
|
if isinstance(rules, list):
|
|
counters = {}
|
|
for item in rules:
|
|
if isinstance(item, dict):
|
|
key = str(item.get("key", "")).strip()
|
|
if key:
|
|
counters[key] = self._safe_int(item.get("value", 0))
|
|
rules = counters
|
|
if not isinstance(rules, dict):
|
|
rules = {}
|
|
|
|
counters = self._format_btwaf_rule_counters(rules)
|
|
total = self._safe_int(data.get("total", 0)) if isinstance(data, dict) else 0
|
|
if total <= 0:
|
|
total = sum(counters.values())
|
|
|
|
return total, counters
|
|
|
|
def _btwaf_attack_types_from_overview(self, overview):
|
|
raw_counters = {}
|
|
rows = overview.get("type", []) if isinstance(overview, dict) else []
|
|
for row in rows if isinstance(rows, list) else []:
|
|
key = ""
|
|
value = 0
|
|
if isinstance(row, dict):
|
|
key = str(row.get("key", row.get("type", row.get("name", "")))).strip()
|
|
value = self._safe_int(row.get("value", row.get("count", 0)))
|
|
elif isinstance(row, (list, tuple)) and len(row) >= 2:
|
|
key = str(row[0]).strip()
|
|
value = self._safe_int(row[1])
|
|
if key:
|
|
raw_counters[key] = value
|
|
if not raw_counters:
|
|
return [], {}
|
|
counters = dict((key, 0) for key in self.WAF_INTERCEPT_KEYS)
|
|
for key, value in raw_counters.items():
|
|
counters[key] = value
|
|
items = [{"type": key, "count": value} for key, value in raw_counters.items() if value > 0]
|
|
items.sort(key=lambda item: item["count"], reverse=True)
|
|
return items, counters
|
|
|
|
def _format_btwaf_rule_counters(self, rules):
|
|
raw = {}
|
|
for key, value in rules.items():
|
|
raw[str(key)] = self._safe_int(value)
|
|
raw["get"] = 0
|
|
if "args" in raw:
|
|
raw["get"] += raw.pop("args")
|
|
if "url" in raw:
|
|
raw["get"] += raw.pop("url")
|
|
|
|
result = {}
|
|
for key in self.WAF_INTERCEPT_KEYS:
|
|
result[key] = self._safe_int(raw.get(key, 0))
|
|
return result
|
|
|
|
def _btwaf_site_config(self, plugin_dir):
|
|
server_file = os.path.join("/www/server/btwaf", "site_waf_config_php.json")
|
|
plugin_file = os.path.join(plugin_dir, "site_waf_config_php.json")
|
|
data = self._read_json(server_file, None)
|
|
if data is None:
|
|
data = self._read_json(plugin_file, {})
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
def _btwaf_intercept_types(self, plugin_dir):
|
|
counters = dict((key, 0) for key in self.WAF_INTERCEPT_KEYS)
|
|
data = self._btwaf_site_config(plugin_dir)
|
|
rows = data.get("data", []) if isinstance(data.get("data", []), list) else []
|
|
for row in rows:
|
|
if not isinstance(row, dict):
|
|
continue
|
|
for item in row.get("total", []) if isinstance(row.get("total", []), list) else []:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
key = str(item.get("key", "")).strip()
|
|
if not key:
|
|
continue
|
|
if key not in counters:
|
|
counters[key] = 0
|
|
counters[key] += self._safe_int(item.get("value", 0))
|
|
|
|
if sum(counters.values()) == 0:
|
|
sqlite_counters = self._btwaf_intercept_types_from_sqlite()
|
|
for key, value in sqlite_counters.items():
|
|
if key not in counters:
|
|
counters[key] = 0
|
|
counters[key] += value
|
|
return counters
|
|
|
|
def _btwaf_intercept_types_from_sqlite(self):
|
|
counters = {}
|
|
db_file = "/www/server/btwaf/totla_db/totla_db.db"
|
|
if not os.path.exists(db_file):
|
|
return counters
|
|
try:
|
|
import sqlite3
|
|
conn = sqlite3.connect(db_file)
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"select coalesce(nullif(value_risk, ''), filter_rule) as risk_type, count(1) "
|
|
"from totla_log group by coalesce(nullif(value_risk, ''), filter_rule)"
|
|
)
|
|
for key, value in cur.fetchall():
|
|
key = str(key or "").strip()
|
|
if key:
|
|
counters[key] = self._safe_int(value)
|
|
conn.close()
|
|
except Exception:
|
|
pass
|
|
return counters
|
|
|
|
def _get_fail2ban_service_status(self, plugin_dir):
|
|
try:
|
|
status = public.run_plugin("fail2ban", "get_fail2ban_status", public.to_dict_obj({}))
|
|
if isinstance(status, bool):
|
|
return status
|
|
if isinstance(status, dict):
|
|
if "status" in status:
|
|
return bool(status.get("status"))
|
|
if "message" in status:
|
|
return bool(status.get("message"))
|
|
except Exception:
|
|
pass
|
|
return os.path.exists(os.path.join(plugin_dir, "fail2ban.sock"))
|
|
|
|
def _top_counter_text(self, counters, limit=3):
|
|
pairs = [
|
|
(key, self._safe_int(value))
|
|
for key, value in counters.items()
|
|
if self._safe_int(value) > 0
|
|
]
|
|
pairs.sort(key=lambda item: item[1], reverse=True)
|
|
if not pairs:
|
|
return "no classified attacks"
|
|
return ", ".join("{} {}".format(key, value) for key, value in pairs[:limit])
|
|
|
|
def _get_protect_days(self):
|
|
install_file = self._panel_file("data", "safeCloud", "install_time.pl")
|
|
ts = self._to_timestamp(self._read_text(install_file).strip())
|
|
if not ts:
|
|
return 1
|
|
days = int((time.time() - ts) / 86400)
|
|
return days if days > 0 else 1
|
|
|
|
def _get_virus_update_time(self):
|
|
rule_dir = self._panel_file("data", "safeCloud", "rules")
|
|
latest = 0
|
|
if os.path.exists(rule_dir):
|
|
for root, dirs, files in os.walk(rule_dir):
|
|
for filename in files:
|
|
try:
|
|
latest = max(latest, int(os.path.getmtime(os.path.join(root, filename))))
|
|
except Exception:
|
|
pass
|
|
return self._format_time(latest)
|
|
|
|
def _ssh_option_enabled(self, conf, option, default=False):
|
|
if not conf:
|
|
return default
|
|
option = str(option or "").lower()
|
|
for line in conf.splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
parts = line.split()
|
|
if len(parts) < 2 or parts[0].lower() != option:
|
|
continue
|
|
value = str(parts[1]).lower()
|
|
if value in ["yes", "true", "1", "on"]:
|
|
return True
|
|
if value in ["no", "false", "0", "off"]:
|
|
return False
|
|
return default
|
|
return default
|
|
|
|
def _panel_file(self, *parts):
|
|
path = os.path.join(self.panel_path, *parts)
|
|
if os.path.exists(path):
|
|
return path
|
|
fallback = os.path.join("/www/server/panel", *parts)
|
|
return fallback
|
|
|
|
def _plugin_path(self, plugin_name):
|
|
path = os.path.join(self.panel_path, "plugin", plugin_name)
|
|
if os.path.exists(path):
|
|
return path
|
|
return os.path.join("/www/server/panel/plugin", plugin_name)
|
|
|
|
def _read_json(self, path, default):
|
|
try:
|
|
body = self._read_text(path)
|
|
if not body:
|
|
return default
|
|
return json.loads(body)
|
|
except Exception:
|
|
return default
|
|
|
|
def _read_text(self, path):
|
|
try:
|
|
if not path or not os.path.exists(path):
|
|
return ""
|
|
with open(path, "r", encoding="utf-8", errors="ignore") as fp:
|
|
return fp.read()
|
|
except TypeError:
|
|
try:
|
|
with open(path, "r") as fp:
|
|
return fp.read()
|
|
except Exception:
|
|
return ""
|
|
except Exception:
|
|
return ""
|
|
|
|
def _tail_lines(self, path, limit):
|
|
if not path or not os.path.exists(path):
|
|
return []
|
|
try:
|
|
with open(path, "r", encoding="utf-8", errors="ignore") as fp:
|
|
return list(deque(fp, maxlen=limit))
|
|
except TypeError:
|
|
try:
|
|
with open(path, "r") as fp:
|
|
return list(deque(fp, maxlen=limit))
|
|
except Exception:
|
|
return []
|
|
except Exception:
|
|
return []
|
|
|
|
def _iter_lines(self, path):
|
|
if not path or not os.path.exists(path):
|
|
return []
|
|
try:
|
|
with open(path, "r", encoding="utf-8", errors="ignore") as fp:
|
|
return list(fp)
|
|
except TypeError:
|
|
try:
|
|
with open(path, "r") as fp:
|
|
return list(fp)
|
|
except Exception:
|
|
return []
|
|
except Exception:
|
|
return []
|
|
|
|
def _unwrap_message(self, result, default):
|
|
try:
|
|
if isinstance(result, dict):
|
|
return result.get("message", default)
|
|
except Exception:
|
|
pass
|
|
return default
|
|
|
|
def _severity_counts(self, include_info=False):
|
|
data = {
|
|
"critical": 0,
|
|
"high": 0,
|
|
"medium": 0,
|
|
"low": 0,
|
|
}
|
|
if include_info:
|
|
data["info"] = 0
|
|
return data
|
|
|
|
def _max_severity(self, counts):
|
|
for key in ["critical", "high", "medium", "low"]:
|
|
if int(counts.get(key, 0) or 0) > 0:
|
|
return key
|
|
return "info"
|
|
|
|
def _normalize_level(self, value):
|
|
if value is None:
|
|
return "", 0
|
|
try:
|
|
num = int(value)
|
|
if num >= 4:
|
|
return "critical", 4
|
|
if num == 3:
|
|
return "high", 3
|
|
if num == 2:
|
|
return "medium", 2
|
|
if num == 1:
|
|
return "low", 1
|
|
return "info", 0
|
|
except Exception:
|
|
pass
|
|
text = str(value).strip().lower()
|
|
if text in ["serious", "critical", "danger", "dangerous", "fatal"]:
|
|
return "critical", 4
|
|
if text in ["high", "important"]:
|
|
return "high", 3
|
|
if text in ["medium", "warning", "warn"]:
|
|
return "medium", 2
|
|
if text in ["low"]:
|
|
return "low", 1
|
|
if text in ["info", "normal", "safe"]:
|
|
return "info", 0
|
|
return "", 0
|
|
|
|
def _to_timestamp(self, value):
|
|
if value is None or value == "":
|
|
return 0
|
|
try:
|
|
ts = int(float(value))
|
|
if ts > 0:
|
|
return ts
|
|
except Exception:
|
|
pass
|
|
text = str(value).strip()
|
|
if not text:
|
|
return 0
|
|
text = text.replace("/", "-")
|
|
for fmt in [
|
|
"%Y-%m-%d %H:%M:%S",
|
|
"%Y-%m-%d %H:%M",
|
|
"%Y-%m-%d",
|
|
"%b %d %H:%M:%S",
|
|
"%b %d %H:%M:%S",
|
|
]:
|
|
try:
|
|
if fmt.startswith("%b"):
|
|
text2 = "{} {}".format(time.localtime().tm_year, text)
|
|
return int(time.mktime(time.strptime(text2, "%Y " + fmt)))
|
|
return int(time.mktime(time.strptime(text, fmt)))
|
|
except Exception:
|
|
continue
|
|
return 0
|
|
|
|
def _parse_syslog_time(self, line):
|
|
text = " ".join(line.split()[:3])
|
|
return self._to_timestamp(text) or int(time.time())
|
|
|
|
def _format_time(self, ts):
|
|
ts = self._to_timestamp(ts)
|
|
if not ts:
|
|
return ""
|
|
try:
|
|
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts))
|
|
except Exception:
|
|
return ""
|
|
|
|
def _relative_time(self, ts):
|
|
ts = self._to_timestamp(ts)
|
|
if not ts:
|
|
return "No data"
|
|
diff = int(time.time()) - ts
|
|
if diff < 60:
|
|
return "Just now"
|
|
if diff < 3600:
|
|
return "{} minutes ago".format(max(1, int(diff / 60)))
|
|
if diff < 86400:
|
|
return "{} hours ago".format(max(1, int(diff / 3600)))
|
|
return "{} days ago".format(max(1, int(diff / 86400)))
|
|
|
|
def _extract_ip(self, text):
|
|
match = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", text or "")
|
|
return match.group(1) if match else ""
|
|
|
|
def _safe_int(self, value, default=0):
|
|
try:
|
|
return int(float(value))
|
|
except Exception:
|
|
return default
|
|
|
|
def _to_bool(self, value, default=False):
|
|
if isinstance(value, bool):
|
|
return value
|
|
if value is None:
|
|
return default
|
|
if isinstance(value, int):
|
|
return value != 0
|
|
text = str(value).strip().lower()
|
|
if text in ["1", "true", "yes", "on", "open"]:
|
|
return True
|
|
if text in ["0", "false", "no", "off", "close", "closed", ""]:
|
|
return False
|
|
return default
|
|
|
|
def _get_int(self, get, name, default, min_value=None, max_value=None):
|
|
value = self._safe_int(self._get_arg(get, name, default), default)
|
|
if min_value is not None:
|
|
value = max(min_value, value)
|
|
if max_value is not None:
|
|
value = min(max_value, value)
|
|
return value
|
|
|
|
def _get_bool(self, get, name, default=False):
|
|
value = self._get_arg(get, name, default)
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, int):
|
|
return value != 0
|
|
return str(value).lower() in ["1", "true", "yes", "on"]
|
|
|
|
def _get_arg(self, get, name, default=None):
|
|
if get is None:
|
|
return default
|
|
args = self._get_request_args(get)
|
|
if name in args:
|
|
return args.get(name, default)
|
|
return default
|
|
|
|
def _get_first_arg(self, get, names, default=None):
|
|
args = self._get_request_args(get)
|
|
for name in names:
|
|
if name in args:
|
|
return args.get(name)
|
|
return default
|
|
|
|
def _get_request_args(self, get):
|
|
args = dict(self._object_items(get))
|
|
try:
|
|
from flask import request
|
|
args.update(request.args.to_dict())
|
|
args.update(request.form.to_dict())
|
|
body = request.get_json(silent=True)
|
|
if isinstance(body, dict):
|
|
args.update(body)
|
|
except Exception:
|
|
pass
|
|
|
|
for key in ["data", "params"]:
|
|
args.update(self._parse_nested_args(args.get(key)))
|
|
return args
|
|
|
|
def _object_items(self, value):
|
|
if isinstance(value, dict):
|
|
return value
|
|
try:
|
|
items = value.get_items()
|
|
return items if isinstance(items, dict) else {}
|
|
except Exception:
|
|
pass
|
|
try:
|
|
items = value.__dict__
|
|
return items if isinstance(items, dict) else {}
|
|
except Exception:
|
|
pass
|
|
return {}
|
|
|
|
def _parse_nested_args(self, value):
|
|
if isinstance(value, dict):
|
|
return value
|
|
items = self._object_items(value)
|
|
if items:
|
|
return items
|
|
if not isinstance(value, str):
|
|
return {}
|
|
value = value.strip()
|
|
if not value:
|
|
return {}
|
|
try:
|
|
parsed = json.loads(value)
|
|
except Exception:
|
|
return {}
|
|
if isinstance(parsed, dict):
|
|
return parsed
|
|
return {}
|