# coding: utf-8 # ------------------------------------------------------------------- # YakPanel # ------------------------------------------------------------------- # Copyright (c) 2014-2099 YakPanel(www.yakpanel.com) All rights reserved. # ------------------------------------------------------------------- # Author: yakpanel # ------------------------------------------------------------------- # ------------------------------ # server safe app # ------------------------------ import json import os import re from copy import deepcopy from typing import Callable import public from public.exceptions import HintException from public.validate import Param public.sys_path_append("class_v2/") from ssh_security_v2 import ssh_security from config_v2 import config class main: def __init__(self): # {name:安全项名称,desc:描述, # suggest:修复建议,check:检查函数,repair:修复函数,value:获取当前值函数,status:状态} self.config = [ { "name": "Default SSH Port", "desc": public.lang("Modify the default SSH port to improve server security"), "suggest": public.lang("Use a high port other than 22"), "check": self.check_ssh_port, }, { "name": "Password Complexity Policy", "desc": public.lang("Enable password complexity check to ensure password security"), "suggest": public.lang("Use a level greater than 3"), "check": self.check_ssh_minclass, "repair": self.repair_ssh_minclass, }, { "name": "Password Length Limit", "desc": public.lang("Set minimum password length requirement"), "suggest": public.lang("Use a password of 9-20 characters"), "check": self.check_ssh_security, "repair": self.repair_ssh_passwd_len, }, { "name": "SSH Login Alert", "desc": public.lang("Send alert notification upon SSH login"), "suggest": public.lang("Enable SSH login alert"), "check": self.check_ssh_login_sender, }, { "name": "Root Login Settings", "desc": public.lang("It is recommended to allow key-based login only"), "suggest": public.lang("Allow only SSH key-based login"), "check": self.check_ssh_login_root_with_key, }, { "name": "SSH Brute-force", "desc": public.lang("Prevent SSH brute-force attacks"), "suggest": public.lang("Enable SSH brute-force protection"), "check": self.check_ssh_fail2ban_brute, }, { "name": "Panel Login Alert", "desc": public.lang("Send alert notification upon panel login"), "suggest": public.lang("Enable panel login alert"), "check": self.check_panel_swing, }, { "name": "Panel Google Authenticator login", "desc": public.lang("Enable TOTP for enhanced security"), "suggest": public.lang("Enable OTP authentication"), "check": self.check_panel_login_2fa, }, { "name": "UnAuth Response Status Code", "desc": public.lang("Set the HTTP response status code for unauthenticated access"), "suggest": public.lang("Set 404 as the response code"), "check": self.check_panel_not_auth_code, }, { "name": "Panel SSL", "desc": public.lang("Enable HTTPS encrypted transmission (after setting will restart the panel)"), "suggest": public.lang("Enable panel HTTPS"), "check": self.check_panel_ssl, } ] self.ssh_security_obj = ssh_security() self.config_obj = config() def get_security_info(self, get=None): """ 获取安全评分 """ new_list = deepcopy(self.config) for idx, module in enumerate(new_list): if isinstance(module.get("check"), Callable): try: module["id"] = int(idx) + 1 check_status = module["check"]() module["status"] = check_status.get("status", False) module["value"] = check_status.get("value") except: module["status"] = False module["value"] = None if "check" in module and isinstance(module["check"], Callable): del module["check"] if "repair" in module and isinstance(module["repair"], Callable): del module["repair"] if "value" not in module: module["value"] = None total_score = 100 # 总分 score = total_score / len(new_list) # 每条的分数 missing_count = 0 # 缺少的条数 for module in new_list: if module["status"] is False: missing_count += 1 # 计算总分 security_score = total_score - (missing_count * score) security_score = round(security_score, 2) # 计算得分文本 if security_score >= 90: score_text = public.lang("Secure") elif security_score >= 70: score_text = public.lang("Relatively Secure") elif security_score >= 50: score_text = public.lang("Average Security") else: score_text = public.lang("Insecure") public.set_module_logs("server_secury", "get_security_info", 1) return public.success_v2({ "security_data": new_list, "total_score": total_score, "score_text": score_text, "score": int(security_score) }) def install_fail2ban(self, get): from panel_plugin_v2 import panelPlugin public.set_module_logs("server_secury", "install_fail2ban", 1) return panelPlugin().install_plugin(get) def repair_security(self, get): """ @name 修复安全项 @parma {"name":"","args":{}} """ try: get.validate([ Param("name").String().Require(), Param("args").Dict().Require(), ], [public.validate.trim_filter()]) except Exception as ex: public.print_log("error info: {}".format(ex)) return public.fail_v2(str(ex)) for security in self.config: if security.get("name") == get.name and isinstance(security.get("repair"), Callable): return security["repair"](public.to_dict_obj(get.args)) raise HintException(public.lang(f"Security Repair Item [{get.name}] Not Found!")) @staticmethod def _find_pwquality_conf_with_keyword(re_search: str) -> str: """ 读取ssh密码复杂度配置 @param re_search: 正则表达式 """ try: if not re_search: raise HintException("required parameter re_search") p_file = '/etc/security/pwquality.conf' p_body = public.readFile(p_file) if not p_body: return "" # 无配置文件时 tmp = re.findall(re_search, p_body, re.M) if not tmp: return "" # 未设置minclass find = tmp[0].strip() return find except: return "" # 异常时认为无 # =================== 检查函数 =================== def check_ssh_port(self) -> dict: """ @name 检查SSH端口是否为默认端口22 """ current_port = public.get_ssh_port() return {"status": current_port != 22, "value": current_port} def check_ssh_minclass(self) -> dict: """ @name 检查SSH密码复杂度策略 """ re_pattern = r"\n\s*minclass\s+=\s+(.+)" find = self._find_pwquality_conf_with_keyword(re_pattern) if not find: return {"status": False, "value": None} # 未设置minclass minclass_value = int(find) return {"status": minclass_value >= 3, "value": minclass_value} def check_ssh_security(self) -> dict: """ @name 检查SSH密码长度限制 """ re_pattern = r"\s*minlen\s+=\s+(.+)" find = self._find_pwquality_conf_with_keyword(re_pattern) if not find: return {"status": True, "value": None} # 未设置minlen时认为无风险 minlen_value = int(find) return {"status": minlen_value >= 9, "value": minlen_value} def check_panel_swing(self) -> dict: """ @name 检查面板登录告警是否开启 """ tip_files = [ "panel_login_send.pl", "login_send_type.pl", "login_send_mail.pl", "login_send_dingding.pl" ] enabled_files = [] for fname in tip_files: filename = "data/" + fname if os.path.exists(filename): enabled_files.append(fname) break is_enabled = len(enabled_files) > 0 value = None if not is_enabled: return {"status": False, "value": value} task_file_path = "/www/server/panel/data/mod_push_data/task.json" sender_file_path = "/www/server/panel/data/mod_push_data/sender.json" task_data = {} try: with open(task_file_path, "r") as file: tasks = json.load(file) # 读取发送者配置文件 with open(sender_file_path, "r") as file: senders = json.load(file) sender_dict = { sender["id"]: sender for sender in senders } # 查找特定的告警任务 for task in tasks: if task.get("keyword") == "panel_login": task_data = task sender_types = set() # 使用集合来保证类型的唯一性 # 对应sender的ID，获取sender_type，并保证唯一性 for sender_id in task.get("sender", []): if sender_id in sender_dict: sender_types.add(sender_dict[sender_id]["sender_type"]) # 将唯一的通道类型列表转回列表格式，添加到告警数据中 task_data["channels"] = list(sender_types) break except: pass value = task_data return {"status": value.get("status", False), "value": value} def check_ssh_login_sender(self) -> dict: """ @name 检查SSH登录告警是否启用 """ result = self.ssh_security_obj.get_login_send(None) res = public.find_value_by_key( result, "result", "error" ) return {"status": res != "error", "value": res} def check_ssh_login_root_with_key(self) -> dict: """ @name 检查SSH是否仅允许密钥登录root """ parsed = self.ssh_security_obj.paser_root_login() current_policy = None try: current_policy = parsed[1] except Exception as e: import traceback public.print_log("error info: {}".format(traceback.format_exc())) return {"status": current_policy == "without-password", "value": current_policy} def check_ssh_fail2ban_brute(self) -> dict: """ @name 检查SSH防爆破是否启用 """ from safeModelV2.sshModel import main as sshmod cfg = sshmod._get_ssh_fail2ban() or {} current_value = cfg.get("status", 0) return {"status": current_value == 1, "value": current_value} def check_panel_login_2fa(self) -> dict: """ @name 检查面板登录动态口令认证是否启用 """ current_value = self.config_obj.check_two_step(None) res = public.find_value_by_key( current_value, "result", False ) return {"status": bool(res), "value": res} def check_panel_not_auth_code(self) -> dict: """ @name 检查面板未登录响应状态码是否设置为 400+ """ current_code = self.config_obj.get_not_auth_status() return {"status": current_code != 0, "value": current_code} def check_panel_ssl(self): """ @name 检查面板是否开启SSL """ enabled = os.path.exists("data/ssl.pl") return {"status": bool(enabled), "value": enabled} # =================== 修复函数 =================== def repair_ssh_minclass(self, get): """ @name 修复SSH密码复杂度 @param {"minclass":9} """ try: get.validate([ Param("minclass").Integer(">", 0).Require(), ], [public.validate.trim_filter()]) except Exception as ex: public.print_log("error info: {}".format(ex)) return public.fail_v2(str(ex)) minclass = int(get.minclass) file = "/etc/security/pwquality.conf" result = { "status": False, "msg": public.lang("Failed to set SSH password complexity, " "please disable system hardening or set it manually") } if not os.path.exists(file): public.ExecShell("apt install libpam-pwquality -y") if os.path.exists(file): f_data = public.readFile(file) if re.findall("\n\s*minclass\s*=\s*\d*", f_data): file_result = re.sub("\n\s*minclass\s*=\s*\d*", "\nminclass = {}".format(minclass), f_data) else: file_result = f_data + "\nminclass = {}".format(minclass) public.writeFile(file, file_result) f_data = public.readFile(file) if f_data.find("minclass = {}".format(minclass)) != -1: result["status"] = True result["msg"] = public.lang("SSH minimum password complexity has been set") return public.return_message(0 if result["status"] else 1, 0, result["msg"]) def repair_ssh_passwd_len(self, get): """ @name SSH密码最小长度设置 @param {"len":9} """ try: get.validate([ Param("len").Integer(">", 0).Require(), ], [public.validate.trim_filter()]) except Exception as ex: public.print_log("error info: {}".format(ex)) return public.fail_v2(str(ex)) pwd_len = int(get.len) file = "/etc/security/pwquality.conf" result = { "status": False, "msg": public.lang("Failed to set SSH minimum password length, please set it manually") } if not os.path.exists(file): public.ExecShell("apt install libpam-pwquality -y") if os.path.exists(file): f_data = public.readFile(file) ssh_minlen = "\n#?\s*minlen\s*=\s*\d*" file_result = re.sub(ssh_minlen, "\nminlen = {}".format(pwd_len), f_data) public.writeFile(file, file_result) f_data = public.readFile(file) if f_data.find("minlen = {}".format(pwd_len)) != -1: result["status"] = True result["msg"] = "SSH minimum password length has been set to {}".format(pwd_len) return public.return_message(0 if result["status"] else 1, 0, result["msg"])