赞
踩
#!/usr/bin/env python3 # -*- coding:utf8 -*- # Author: # Description: 安全基线检测 import os import re import json import argparse import datetime import subprocess class SafeBaseline: @staticmethod def parameters(): """ 传递参数 :return: """ parser = argparse.ArgumentParser() parser.add_argument("--resultFields", "-resultFields", help="检查项") parser.add_argument("--userWhiteList", "-userWhiteList", help="用户白名单") parser.add_argument("--portWhiteList", "-portWhiteList", help="端口白名单") parser.add_argument("--commandWhiteList", "-commandWhiteList", help="命令白名单") parser.add_argument("--systemWhiteList", "-systemWhiteList", help="系统白名单") params = parser.parse_args() return params @staticmethod def open_file(filename): """ 读取文件内容 :param filename: 文件名 :return: """ with open(filename) as f: data = f.read() return data @classmethod def system_command(cls, command): """ 执行系统命令 :param command: 命令 :return: 输出结果,报错,执行状态 :param command: :return: """ shell = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) stdout, stderr = shell.communicate() try: return stdout.decode("utf8"), stderr.decode("utf8"), shell.returncode except Exception: return stdout.decode("gbk"), stderr.decode("gbk"), shell.returncode def systemAccountCheck(self): """ 1.系统账户安全检查 :return: """ stdout, stderr, return_code = self.system_command("cat /etc/login.defs |egrep '^PASS_MIN_LEN'") password_length = stdout.replace('PASS_MIN_LEN','').strip() warn_level = [] details = [] password_complexity = re.search('pam_cracklib.so.*?\n',self.open_file('/etc/pam.d/system-auth-ac')) if password_complexity: if re.search(r"dcredit=(-?\d+)", password_complexity.group()): dcredit = re.search(r"dcredit=(-?\d+)", password_complexity.group()).group(1) if int(dcredit.replace('-', '')) >= 2: warn_level.append(1) else: details.append('系统账户密码策略要求最少一个数字,当前个数为{}'.format(dcredit.replace('-', ''))) else: details.append('系统账户密码策略要求最少一个数字') if re.search(r"lcredit=(-?\d+)", password_complexity.group()): lcredit = re.search(r"lcredit=(-?\d+)", password_complexity.group()).group(1) if int(lcredit.replace('-', '')) >= 1: warn_level.append(1) else: details.append('系统账户密码策略要求最少一个小写字母,当前个数为{}'.format(lcredit.replace('-', ''))) else: details.append('系统账户密码策略要求最少一个小写字母') if re.search(r"ucredit=(-?\d+)", password_complexity.group()): ucredit = re.search(r"ucredit=(-?\d+)", password_complexity.group()).group(1) if int(ucredit.replace('-', '')) >= 1: warn_level.append(1) else: details.append('系统账户密码策略要求最少一个大写字母,当前个数为{}'.format(ucredit.replace('-', ''))) else: details.append('系统账户密码策略要求最少一个大写字母,当前未配置') if re.search(r"ocredit=(-?\d+)", password_complexity.group()): ocredit = re.search(r"ocredit=(-?\d+)", password_complexity.group()).group(1) if int(ocredit.replace('-', '')) >= 1: warn_level.append(1) else: details.append('系统账户密码策略要求最少一个特殊字符,当前个数为{}'.format(ocredit.replace('-', ''))) else: details.append('系统账户密码策略要求最少一个特殊字符,当前未配置') if re.search(r"minlen=(-?\d+)", password_complexity.group()): minlen = re.search(r"minlen=(-?\d+)", password_complexity.group()).group(1) if int(minlen.replace('-', '')) >= 8: warn_level.append(1) else: details.append('系统账户密码策略要求密码口令最少8位,当前个数为 {}'.format(minlen.replace('-', ''))) else: details.append('系统账户密码策略要求密码口令最少8位,当前未配置') else: if int(password_length) >= 8: warn_level.append(1) details.append({'Conformity': '系统账户密码策略要求密码口令8位','NonConformity': '系统账户密码复杂度其他项未配置'}) else: details.append('系统账户密码复杂度未设置') if len(warn_level) >= 5: result = 0 elif 4 <= len(warn_level) < 5: result = 1 else: result = 2 return {"result": result, "Details": details} def remoteLoginCheck(self): """ 远程登陆检查 :return: """ result = 0 details = [] today = datetime.date.today() start_month = today.strftime("%b") last_month = today.replace(day=1) - datetime.timedelta(days=1) end_month = last_month.strftime("%b") command = "cat /var/log/secure* |grep -E '^%s|^%s'|egrep 'Accept.*[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}.*port.*'|awk '{print $1,$2,$3,$9,$11}'" % (end_month, start_month) stdout, stderr, return_code = self.system_command(command) if stdout: result = 3 for info in stdout.strip().split('\n'): info_list = info.split(' ') if {'USER': info_list[3], 'IPADDRESS': info_list[4]} not in details: details.append({'USER': info_list[3], 'IPADDRESS': info_list[4]} ) return {"result": result, "Details": details} def opensslVersionCheck(self): """ 检查openssl版本是否高于1.1.1 :return: """ details = [] result = 0 stdout, stderr, return_code = self.system_command('openssl version') if re.search('\d+\.\d+\.\d+',stdout).group(): data = ''.join(re.search('\d+\.\d+\.\d+',stdout).group().split('.')) if int(data) > 111: result = 0 else: result = 2 details.append('当前Openssl版本为{}, 安全基线版本要求为1.1.1 '.format('.'.join(data))) return {"result": result, "Details": details} def opensshVersionCheck(self): """ 检查openssh版本是否高于8.6p1 :return: """ details = [] result = 0 stdout, stderr, return_code = self.system_command('ssh -V') if re.findall('OpenSSH_(.*?),', "{}{}".format(stdout, stderr)): data = re.search('OpenSSH_(.*?),', "{}{}".format(stdout, stderr)).group() version = ''.join(re.findall('\d+', data)) if int(version) <= 861: result = 2 details.append('当前Openssh版本为{},安全基线版本要求为8.6p1'.format(data.replace(',', ''))) return {"result": result, "Details": details} def nonSystemDefaultUsersCheck(self): """ 检查非系统默认用户 :return: """ stdout, stderr, return_code = self.system_command("cat /etc/passwd |awk -F ':' '{print $1}'|grep -Ev 'root|sshd|bin|daemon|adm|lp|sync|shutdown|halt|mail|operator|ftp|nobody|systemd-network|dbus|polkitd|libstoragemgmt|rpc|saned|gluster|saslauth|abrt|chrony|unbound|qemu|sssd|usbmuxd|ntp|gdm|rpcuser|nfsnobody|postfix|tcpdump'") if self.parameters().systemWhiteList: non_system_user = [user for user in stdout.split('\n') if user not in self.parameters().systemWhiteList.split(',') and user != '' ] else: non_system_user = [user for user in stdout.split('\n') if user != '' ] result = 1 if non_system_user else 0 details = non_system_user return {"result": result, "Details": details} def userAuthorityCheck(self): """ 列出高权限的用户和用户组确保UID为0的用户只有root, UID为0的用户为高权限用户,判断是否存在其他高权限用户及用户组 :return: """ details = [] result = 0 stdout, stderr, return_code = self.system_command("cat /etc/sudoers|grep -E -v '^#'|grep 'ALL=(ALL)'") default_user_group = ['root', '%wheel'] if self.parameters().userWhiteList: default_user_group.extend(self.parameters().userWhiteList.split(',')) for user in stdout.strip().split('\n'): if user.split('ALL=(ALL)')[0].replace('\t','') not in default_user_group: if user.split('ALL=(ALL)')[0].startswith('%'): result = 2 details.append({'高权限用户组': '{}'.format(user.split('ALL=(ALL)')[0]).replace('\t','')}) else: result = 2 details.append({'高权限用户': '{}'.format(user.split('ALL=(ALL)')[0]).replace('\t','')}) return {"result": result, "Details": details} def historyCommandCheck(self): """ 5.history文件和命令检查 :return: """ result = 0 details = [] bash_history_file = os.path.join(os.path.expanduser('~'), '.bash_history') stdout, stderr, return_code = self.system_command("cat {}".format(bash_history_file)) serious_level_command = [ '> /dev/sda', 'mv $file /dev/null', '.(){ .|.& };.', 'rm -rf /' '^foo^bar', 'dd if=/dev/random of=/dev/sda', ] warning_level_command = [ 'file->', 'wget url -O- | sh', 'wget', 'curl', 'rm -rf *', 'rm -rf .' ] if self.parameters().commandWhiteList: command_list = [command for command in self.parameters().commandWhiteList.split(',') if command != ''] for command in command_list: if command in serious_level_command: serious_level_command.remove(command) if command in warning_level_command: warning_level_command.remove(command) for command in stdout.split('\n'): for serious_command in serious_level_command: if command.startswith(serious_command): result = 2 if command not in details: details.append(command) for warning_command in warning_level_command: if command.startswith(warning_command): print(command) if result != 2: result = 1 if command not in details: details.append(command) return {"result": result, "Details": details} def systemCommandModifyCheck(self): """ 系统命令修改检查 :return: """ shell_script = """ #!/bin/bash --login shopt expand_aliases shopt -s expand_aliases shopt expand_aliases alias """ result = 0 details = [] with open('alias_script_for_check.sh','w') as f: f.write(shell_script.strip()) stdout, stderr, return_code = self.system_command('chmod +x alias_script_for_check.sh && ./alias_script_for_check.sh |grep -v expand && rm -rf alias_script_for_check.sh') system_default_command = [ "alias cp='cp -i'", "alias egrep='egrep --color=auto'", "alias fgrep='fgrep --color=auto'", "alias grep='grep --color=auto'", "alias l.='ls -d .* --color=auto'", "alias ll='ls -l --color=auto'", "alias ls='ls --color=auto'", "alias mv='mv -i'", "alias rm='rm -i'", "alias which='alias | /usr/bin/which --tty-only --read-alias --show-dot --show-tilde'" ] for alias in stdout.strip().split('\n'): if alias not in system_default_command: result = 1 details.append('{}'.format(alias)) return {"result": result, "Details": details} def sshForceAttackCheck(self): """ SSH爆力破解检查 :return: """ command = """ find /var/log -name 'secure*' -type f | while read line;do awk '/Failed/{print $(NF-3)}' $line;done | awk '{a[$0]++}END{for (j in a) if(a[j] > 20) print j"="a[j]}' | sort -n -t'=' -k 2 """ stdout, stderr, return_code = self.system_command(command) details = [] if stdout: result = 2 details.append(stdout) else: result = 0 return {"result": result, "Details": details} def inetdBackDoorCheck(self): """ ssh文件后门检查 :return: """ command = """ [[ -f "~/.ssh/config" ]] && egrep -i 'ProxyCommand|LocalCommand' ~/.ssh/config """ stdout, stderr, return_code = self.system_command(command) details = [] if stdout: result = 2 details.append(stdout) else: result = 0 return {"result": result, "Details": details} def maliciousFileCheck(self): """ 恶意文件检查 :return: """ malicious_file_list = [ 'ISY.EXE', '2SY.EXE', 'EXERT.exe', 'ld.so.preload', 'libioset.so', 'watchdogs', 'ksoftirqds', 'EXPIORER.com', 'finders.com', 'Logol_exe', 'LSASS.exe', 'mstask.exe', 'popwin.exe', 'smss.exe', 'SQL Slammer', 'MS Blaster' ] details = [] result = 0 for file in malicious_file_list: stdout, stderr, return_code = self.system_command('find /* -type f -name "{}"'.format(file)) if stdout: details.append('{}'.format(file)) result = 2 return {"result": result, "Details": details} def inetdConfBackDoorFileCheck(self): """ /etc/inetd.conf文件后门检查 :return: """ command = """ [[ -f "/etc/inetd.conf" ]] && grep -E '(bash -i)' /etc/inetd.conf """ stdout, stderr, return_code = self.system_command(command) details = [] if stdout: result = 2 details.append(stdout) else: result = 0 return {"result": result, "Details": details} def crontabCheck(self): """ crontab计划检查 :return: """ malicious_script_execution_plan = ['wget', 'cron.hourly'] result = 0 details = [] for plan in malicious_script_execution_plan: stdout, stderr, return_code = self.system_command('crontab -l|grep {}'.format(plan)) if stdout: result = 2 details.append('{}'.format(stdout).replace('\n','')) return {"result": result, "Details": details} def maliciousProcessCheck(self): """ 12.恶意进程检查 :return: """ result = 0 details = [] system_service_default_process_white_list = [ 'uwsgi', 'python', 'kthreadd', 'kworker', 'ksoftirqd/0', 'migration/0', 'rcu_bh', 'rcu_sched', 'lru-add-drain', 'watchdog/0', 'watchdog/1', 'migration/1', 'ksoftirqd/1]', 'kworker/1:0H', 'watchdog/2', 'ksoftirqd/2', 'netns', 'kdevtmpfs]', 'kworker/3:0H]', 'writeback', 'watchdogd', 'ksmd', 'crypto', 'xfs-buf/dm-0', 'xfs-data/dm-0', 'xfs-reclaim/dm-', 'xfs-log/dm-0', 'kworker/2:3', 'systemd-journald', 'systemd-udevd', 'auditd', 'audispd', 'sedispatch', 'vmtoolsd', 'bluetoothd', 'ModemManager', 'rtkit-daemon', 'chronyd', 'accounts-daemon', 'systemd-logind', 'udisksd', 'ksmtuned', 'libvirtd', 'libvirt_leaseshelper', 'upowerd' '/usr/sbin/gdm', '/usr/libexec/boltd', '/usr/libexec/packagekitd', 'wpa_supplicant.pid', '/usr/libexec/colord', 'imsettings-daemon', '/usr/libexec/gvfsd', 'at-spi-bus-launcher', 'gnome-shell', 'ibus-dconf', 'ibus-portal', 'gnome-shell-calendar-server', 'xdg-permission-store', 'evolution-source-registry', 'dconf-service', 'mission-control-5', 'gvfs-udisks2-volume-monitor', 'goa-daemon', 'gvfs-afc-volume-monitor', 'gvfs-gphoto2-volume-monitor', 'goa-identity-service', 'gvfs-mtp-volume-monitor', 'gvfs-goa-volume-monitor', 'gsd-power','gsd-print-notifications', 'gsd-rfkill', 'gsd-screensaver-proxy', 'gsd-sharing', 'gsd-sound', 'gsd-xsettings', 'gsd-wacom', 'gsd-smartcard', 'gsd-account', 'gsd-a11y-settings', 'gsd-clipboard', 'gsd-color', 'gsd-datetime', 'gsd-housekeeping', 'gsd-keyboard', 'evolution-calendar-factory', 'gsd-media-keys', 'gsd-mouse', 'gsd-printer', 'evolution-addressbook-factory', 'gsd-disk-utility-notify', 'tracker-extract', 'tracker-miner-apps', 'tracker-miner-fs', 'tracker-miner-user-guides', 'tracker-store', 'ibus-engine-simple', 'gvfsd-metadata', 'fwupd','gconfd-2', '-bash', 'dhclient', 'abrt-applet', 'awk','systemd', 'sshd', 'ps', 'bash', 'gdm-session-worker', 'gnome', 'sleep', 'NetworkManager', 'rngd', 'rpcbind', 'crond', 'rsyslogd', 'lsmd', 'atd', 'smartd', 'lvmetad', 'dbus-daemon', 'ssh-agent', 'dnsmasq', 'upowerd', 'ibus-daemon', 'avahi-daemon', 'alsactl', 'clickhouse', 'postgres', 'httpd', 'dbus-launch', 'NetworkManager', 'java', ] command = "ps -f --ppid 2 -p 2 -N | grep -v grep|grep -v PID|awk -F ' ' '{print $1,$2,$8}'|grep -Ev '%s'" % '|'.join(system_service_default_process_white_list) # print(command) stdout, stderr, return_code = self.system_command(command) if stdout: result = 1 for info in stdout.strip().split('\n'): try: data = info.split(' ') if {'USER': data[0], 'PID': data[1], 'CMD': data[2]} not in details: details.append({'USER': data[0], 'PID': data[1], 'CMD': data[2]}) except Exception as e: exception = e return {"result": result, "Details": details} def portListenCheck(self): """ 监听端口检查 :return: """ result = 0 details = [] safe_level_port_list = [] product_port = [ '18080-18089', '18093-18096', 18091, '18100-18144', '18160-18165', 123, '18201-18209', '18211-28212', '18216-18217', 18220, 18226, '18241-18242', '18246-18248', '18250-18252', 18256, '18260-18261', 18256, '18260-18261', 18266, 18274, 18281, '18286-18287', '18292-18305', '18311-18312', 18316, '18321-18333', 18336, '18501-18508', '18355-18358', '18341-18344', '18346-18348', '18371-18375', '18377-18380', '18383-18391', 18406, 18408, '18421-18426', '18431-18434', '18436-18486', '18488-18493', '19001-19005', '19011-19030', '20-23', 25, 53, 69, '80-89', 443, '8440-8450', '8080-8089', '110-111', 2049, 137, 139, 445, 143, 161, 389, '512-514', 873, 1194, 1352, 1433, 1521, 1500, 1723, '2082-2083', 2181, 2601, 2604, 3128, '3311-3312', 3306, 3389, 3690, 4848, 5000, 5432, '5900-5902', 5984, 6379, '7001-7002', 7778, 8000, 8443, 8069, '9080-9081', 9090, 9200, 9300, 11211, 27017,27018, 50000, 50070, 50030, 58, 894 ] if self.parameters().portWhiteList: port_white_list = [int(i) for i in self.parameters().portWhiteList.split(',') if i != ''] product_port.extend(port_white_list) for port in product_port: if isinstance(port,str): s_number = int(port.split('-')[0]) e_number = int(port.split('-')[1]) for i in range(s_number, e_number+1): safe_level_port_list.append(i) else: safe_level_port_list.append(port) # command = "netstat -anlp|awk -F ' ' '{print $4,$7}'| grep -v '\['|grep -v 'ACC' |grep -v ']'|awk -F ':' '{print $NF}'|grep -P '\d'|grep '/'" command = " ss -tunlp|grep -v Local|awk '{print $5,$7}'" stdout, stderr, return_code = self.system_command(command) for port in stdout.strip().split('\n'): result = 1 PORT = int(port.split(' ')[0].split(':')[-1]) # ProgramName = re.search('"(.*?)"',port.split(' ')[1]).group().replace('"','') PID = re.search('pid=\d+',port.split(' ')[1]).group().replace('pid=','') cmd = """ awk '{$1=$2=$3=$4=$5=$6=$7=""; print $0}' """ stdout, stderr, return_code = self.system_command("ps -ef |grep {}|grep -v 'ps -ef'|grep -v grep|{}".format(PID,cmd)) ProgramName = stdout.strip().split('\n')[0] if PORT not in safe_level_port_list: if {'PORT':PORT, 'ProgramName':ProgramName, 'PID': PID} not in details: details.append({'PORT':PORT, 'ProgramName':ProgramName, 'PID': PID}) return {"result": result, "Details": details} def miningFileProgressCheck(self): """ 挖矿文件进程检查 :return: """ result = 0 details = [] mining_file = ['ZavD6x','wbew', 'httpdz','lru-add-drain', 'wwatchdog'] for file in mining_file: command = " ps -aux |grep -E '{}'|grep -v grep".format(file) stdout, stderr, return_code = self.system_command(command) if stdout: result = 2 details.append('{}'.format(file)) return {"result": result, "Details": details} def run(self): """ 调用逻辑 :return: """ system_level = ["systemAccountCheck", "remoteLoginCheck", "opensslVersionCheck", "opensshVersionCheck"] users_level = [ "nonSystemDefaultUsersCheck", "userAuthorityCheck", "historyCommandCheck", "systemCommandModifyCheck", "sshForceAttackCheck", "inetdBackDoorCheck" ] file_level = ["maliciousFileCheck", "inetdConfBackDoorFileCheck", "crontabCheck"] process_level = ["maliciousProcessCheck", "portListenCheck"] event_level = ["miningFileProgressCheck"] data = {} result_fields_data = [] if self.parameters().resultFields: result_fields_data = self.parameters().resultFields.split(',') else: result_fields_data.extend(system_level) result_fields_data.extend(users_level) result_fields_data.extend(file_level) result_fields_data.extend(process_level) result_fields_data.extend(event_level) for field in result_fields_data: field_value = eval("self.%s()" % field) if field in system_level: if not data.get("systemLevel"): data["systemLevel"] = {} data["systemLevel"].update({field: field_value}) elif field in users_level: if not data.get("usersLevel"): data["usersLevel"] = {} data["usersLevel"].update({field: field_value}) elif field in file_level: if not data.get("fileLevel"): data["fileLevel"] = {} data["fileLevel"].update({field: field_value}) elif field in process_level: if not data.get("processLevel"): data["processLevel"] = {} data["processLevel"].update({field: field_value}) elif field in event_level: if not data.get("eventLevel"): data["eventLevel"] = {} data["eventLevel"].update({field: field_value}) result_list = [] if data: for level in list(data.keys()): for check in data.get(level): result = data.get(level).get(check).get('result') result_list.append(result) if 2 in result_list: riskLevel = 2 elif 1 in result_list: riskLevel = 1 else: riskLevel = 0 check_result = { "riskLevel": riskLevel, "data": data } print(json.dumps(check_result,ensure_ascii=False)) return json.dumps(check_result,ensure_ascii=False) class Html: def __init__(self): self.safe_baseline = SafeBaseline() self.json_params = json.loads(self.safe_baseline.run()) self.level = [{'key': 2, 'value': '<font color="red">严重</font>'}, {'key': 1, 'value': '<font color="orange">警告</font>'}, {'key': 0, 'value': '<font color="info">安全</font>'}, {'key': 3, 'value': '<font color="blue">人工审核</font>'}] @staticmethod def create_file(filename, html): with open(filename, 'w') as f: f.write(html) @staticmethod def replace(file_name, before, after): with open(file_name, 'r+') as f: t = f.read() t = t.replace(before, after) f.seek(0, 0) f.write(t) f.truncate() @staticmethod def html_body(): message = """ <!DOCTYPE HTML > <html> <head> <meta charset="utf-8"> <title>安全评估检测报告</title> <link rel="stylesheet" href="bootstrap/css/bootstrap.min.css"> <link rel="stylesheet" href="bootstrap/css/bootstrap.css"> </head> <body> <div class="container-fluid"> <div class="page-header"> <center><h1>主机安全评估检测报告</h1></center> </div> <div> <center><h2> 安全检测级别说明</h2></center> <table class="table table-bordered table-striped"> <tr> <th>安全检测级别</th> <th>检测级别说明</th> </tr> <tr> <th width="200">严重级</th> <td width="200">需进行整改</td> </tr> <tr> <th>警告级</th> <td>需根据实际情况选择整改</td> </tr> <tr> <th>人工审计</th> <td>需要人工判断有无风险</td> </tr> <tr> <th>安全级</th> <td>安全级表示主机无风险</td> </tr> </table> </div> {{safeCheckOverview}} {{systemLevel}} {{systemAccountCheck}} {{remoteLoginCheck}} {{opensslVersionCheck}} {{opensshVersionCheck}} {{usersLevel}} {{nonSystemDefaultUsersCheck}} {{userAuthorityCheck}} {{systemCommandModifyCheck}} {{sshForceAttackCheck}} {{inetdBackDoorCheck}} {{historyCommandCheck}} {{fileLevel}} {{crontabCheck}} {{maliciousFileCheck}} {{inetdConfBackDoorFileCheck}} {{processLevel}} {{maliciousProcessCheck}} {{portListenCheck}} {{eventLevel}} {{miningFileProgressCheck}} </div> </body> </html> """ return message def safeCheckOverview(self): # 安全合规检测概览 check_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') ip_address = " ifconfig|grep inet|grep -E -o '[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}'|grep -v -E '1$|0$|^255|255$|127.0.0.1'" ip_address_stdout, stderr, return_code = self.safe_baseline.system_command(ip_address) risk_level = self.json_params.get('riskLevel') overall_risk_level = '' # 检查整体风险等级 for level in self.level: if level.get('key') == risk_level: overall_risk_level = level.get('value') # 整体风险等级个数统计 result_list = [] if self.json_params.get('data'): for level in list(self.json_params.get('data').keys()): # print(level) for check in self.json_params.get('data').get(level): result = self.json_params.get('data').get(level).get(check).get('result') result_list.append(result) # 单个风险等级个数统计 warning_level_number = len([i for i in result_list if i == 1]) serious_level_number = len([i for i in result_list if i == 2]) safe_level_number = len([i for i in result_list if i == 0]) manual_audit_level_number = len([i for i in result_list if i == 3]) message = f""" <center><h2> 安全合规检测概览 </h2></center> <table class="table table-bordered table-striped"> <tr> <th>整体风险等级</th> <td> <b>{overall_risk_level}</b></td> <th> 严重级别</th> <td><font color="red"> {serious_level_number} </font> 个</td> </tr> <tr> <th>任务名称</th> <td>主机安全合规检测</td> <th>警告级别</th> <td><font color="orange"> {warning_level_number} </font> 个</td> </tr> <tr> <th>扫描对象</th> <td>{ip_address_stdout}</td> <th> 人工审核级别</th> <td><font color="blue"> {manual_audit_level_number} </font>个</td> </tr> <tr> <th width="200">扫描时间</th> <td width="200">{check_time}</td> <th width="200">安全级别</th> <td width="200"><font color="info"> {safe_level_number} </font> 个</td> </tr> </table> <center> <h2>安全合规检测内容</h2></center> <hr> """ return message def systemAccountCheck(self): data = self.json_params.get('data').get('systemLevel').get('systemAccountCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') details = ['<li><font color="black">{}</font></li>'.format(i) for i in data.get('Details')] html_tag = ' '.join(details) message = f""" <h4>系统账户安全检查</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> <tr> <td>详情</td> <td> <ul> {html_tag} </ul> </tr> </table> """ return message def remoteLoginCheck(self): data = self.json_params.get('data').get('systemLevel').get('remoteLoginCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') html_tag = data.get('Details') tmp = [] for detail in html_tag: tmp.append( '<tr> <td width="150">用户名</td> <td width="150">{}</td> <td width="150">IP地址</td> <td width="150"> {} </td> </tr>'.format( detail.get('USER'), detail.get('IPADDRESS'))) message = f""" <h4> 远程登录检查</h4> <table class="table table-striped table-bordered"> <tr> <th colspan="2" width="200">风险等级</th> <th colspan="2" width="200">{safe_check_level}</th> </tr> {''.join(tmp)} </table> """ return message def opensslVersionCheck(self): data = self.json_params.get('data').get('systemLevel').get('opensslVersionCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') # html_tag = ['<tr> <td>详情</td> <td> <font color="black">{}</font></td> </tr>'.format(i) for i in data.get('Details')] message = f""" <h4>openssl版本检查</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> {''.join(html_tag)} </table> """ return message def opensshVersionCheck(self): data = self.json_params.get('data').get('systemLevel').get('opensshVersionCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') # html_tag = ['<tr> <td>详情</td> <td> <font color="black">{}</font></td> </tr>'.format(i) for i in data.get('Details')] message = f""" <h4>openssh版本检查</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> {''.join(html_tag)} </table> """ return message def nonSystemDefaultUsersCheck(self): data = self.json_params.get('data').get('usersLevel').get('nonSystemDefaultUsersCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') details = ['<li><font color="black">{}</font></li>'.format(i) for i in data.get('Details')] html_tag = ' '.join(details) message = f""" <h4>非系统默认用户检测</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> <tr> <td>人工审核用户列表</td> <td> <ul> {html_tag} </ul> </td> </tr> </table> """ return message def userAuthorityCheck(self): data = self.json_params.get('data').get('usersLevel').get('userAuthorityCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') details = ['<li><font color="black">{}</font></li>'.format(i) for i in data.get('Details')] html_tag = ' '.join(details) message = f""" <h4>高权限的用户和用户组检测</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> <tr> <td>人工审核高权限的用户和用户组</td> <td> <ul> {html_tag} </ul> </td> </tr> </table> """ return message def systemCommandModifyCheck(self): data = self.json_params.get('data').get('usersLevel').get('systemCommandModifyCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') details = ['<li><font color="black">{}</font></li>'.format(i) for i in data.get('Details')] html_tag = ' '.join(details) message = f""" <h4>系统命令被修改的内容和被修改时间检查</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> <tr> <td>人工审核命令修改记录</td> <td> <ul> {html_tag} </ul> </td> </tr> </table> """ return message def sshForceAttackCheck(self): data = self.json_params.get('data').get('usersLevel').get('sshForceAttackCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') details = ['<li><font color="black">{}</font></li>'.format(i) for i in data.get('Details')] html_tag = ' '.join(details) message = f""" <h4>SSH爆力破解检查</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> <tr> <td>详情</td> <td> <ul> {html_tag} </ul> </td> </tr> </table> """ return message def inetdBackDoorCheck(self): data = self.json_params.get('data').get('usersLevel').get('inetdBackDoorCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') details = ['<li><font color="black">{}</font></li>'.format(i) for i in data.get('Details')] html_tag = ' '.join(details) message = f""" <h4>SSH 后门配置/inetd后门检查</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> <tr> <td>详情</td> <td> <ul> {html_tag} </ul> </td> </tr> </table> """ return message def historyCommandCheck(self): data = self.json_params.get('data').get('usersLevel').get('historyCommandCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') details = ['<li><font color="black">{}</font></li>'.format(i) for i in data.get('Details')] html_tag = ' '.join(details) message = f""" <h4>history文件和命令检查</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> <tr> <td>人工审核历史命令</td> <td> <ul> {html_tag} </ul> </td> </tr> </table> """ return message def maliciousFileCheck(self): data = self.json_params.get('data').get('fileLevel').get('maliciousFileCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') details = ['<li><font color="black">{}</font></li>'.format(i) for i in data.get('Details')] html_tag = ' '.join(details) message = f""" <h4>恶意文件检查</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> <tr> <td> 详情</td> <td> <ul> {html_tag} </ul> </td> </tr> </table> """ return message def inetdConfBackDoorFileCheck(self): data = self.json_params.get('data').get('fileLevel').get('inetdConfBackDoorFileCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') details = ['<li><font color="black">{}</font></li>'.format(i) for i in data.get('Details')] html_tag = ' '.join(details) message = f""" <h4>/etc/inetd.conf文件后门检查</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> <tr> <td> 详情</td> <td> <ul> {html_tag} </ul> </td> </tr> </table> """ return message def crontabCheck(self): data = self.json_params.get('data').get('fileLevel').get('crontabCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') details = ['<li><font color="black">{}</font></li>'.format(i) for i in data.get('Details')] html_tag = ' '.join(details) message = f""" <h4>crontab计划检查</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> <tr> <td> 详情</td> <td> <ul> {html_tag} </ul> </td> </tr> </table> """ return message def maliciousProcessCheck(self): data = self.json_params.get('data').get('processLevel').get('maliciousProcessCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') html_tag = [] for i in data.get('Details'): html_tag.append( '<tr><td width="150">进程命令</td> <td width="150"> {} </td> <td width="150">用户名</td> <td width="150">{}</td> <td width="150">进程ID</td> <td width="150">{}</td> </tr>'.format( i.get('CMD'), i.get('USER'), i.get('PID'))) html_tag = ' '.join(html_tag) message = f""" <h4> 恶意进程检查</h4> <table class="table table-striped table-bordered"> <tr> <th colspan="3" width="200">风险等级</th> <th colspan="3" width="200">{safe_check_level}</th> </tr> {html_tag} </table> """ return message def portListenCheck(self): data = self.json_params.get('data').get('processLevel').get('portListenCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') html_tag = [] for i in data.get('Details'): html_tag.append( '<tr><td width="150">程序名称</td> <td width="150"> {} </td> <td width="150">端口</td> <td width="150">{}</td> <td width="150">进程ID</td> <td width="150">{}</td> </tr>'.format( i.get('ProgramName'), i.get('PORT'), i.get('PID'))) html_tag = ' '.join(html_tag) message = f""" <h4>端口监听检测 </h4> <table class="table table-striped table-bordered"> <tr> <th colspan="3" width="200">风险等级</th> <th colspan="3" width="200">{safe_check_level}</th> </tr> {html_tag} </table> """ return message def miningFileProgressCheck(self): data = self.json_params.get('data').get('eventLevel').get('miningFileProgressCheck') safe_check_level = '' for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') for level in self.level: if level.get('key') == data.get('result'): safe_check_level = level.get('value') details = ['<li><font color="black">{}</font></li>'.format(i) for i in data.get('Details')] html_tag = ' '.join(details) message = f""" <h4>挖矿文件/进程检查</h4> <table class="table table-striped table-bordered"> <tr> <th width="200">风险等级</th> <th width="200">{safe_check_level}</th> </tr> <tr> <td>详情</td> <td> <ul> {html_tag} </ul> </td> </tr> </table> """ return message def systemLevel(self): return "<center><h3>系统级安全检测</h3></center>" def usersLevel(self): return "<center><h3>用户级安全检测</h3></center>" def fileLevel(self): return "<center><h3>文件级安全检测</h3></center>" def processLevel(self): return "<center><h3>进程级安全检测</h3></center>" def eventLevel(self): return "<center><h3>事件级安全检测</h3></center>" def create(self): # 设置文件名及创建html主体结构 filename = '{}.html'.format('主机安全评估检测报告') self.create_file(filename, self.html_body()) # 安全合规检测概览 self.replace(filename, '{{safeCheckOverview}}', self.safeCheckOverview()) # 调用html生成函数 data = self.json_params.get('data') for level in list(data.keys()): self.replace(filename, '{{%s}}' % level, eval("self.%s()" % level)) for key in list(data.get(level).keys()): self.replace(filename, '{{%s}}' % key, eval("self.%s()" % key)) # 清空变量 clear_variable = [ "{{safeCheckOverview}}", "{{systemLevel}}", "{{systemAccountCheck}}", "{{remoteLoginCheck}}", "{{opensslVersionCheck}}", "{{opensshVersionCheck}}", "{{usersLevel}}", "{{nonSystemDefaultUsersCheck}}", "{{userAuthorityCheck}}", "{{systemCommandModifyCheck}}", "{{sshForceAttackCheck}}", "{{inetdBackDoorCheck}}", "{{historyCommandCheck}}", "{{fileLevel}}", "{{crontabCheck}}", "{{eventLevel}}", "{{maliciousFileCheck}}", "{{inetdConfBackDoorFileCheck}}", "{{processLevel}}", "{{maliciousProcessCheck}}", "{{portListenCheck}}", "{{miningFileProgressCheck}}", ] for clear in clear_variable: self.replace(filename, clear, '') if __name__ == '__main__': html = Html() html.create() # python3 test.py --resultFields systemAccountCheck,remoteLoginCheck,opensslVersionCheck,opensshVersionCheck,nonSystemDefaultUsersCheck,userAuthorityCheck,systemCommandModifyCheck,sshForceAttackCheck,inetdBackDoorCheck,maliciousFileCheck,inetdConfBackDoorFileCheck,crontabCheck,maliciousProcessCheck,portListenCheck,miningFileProgressCheck,historyCommandCheck --userWhiteList 'test qwe',wangze --portWhiteList 123,332 --systemWhiteList wqe --commandWhiteList 'abc 2',rr
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。