202 lines
5.7 KiB
Python
202 lines
5.7 KiB
Python
from time import strftime, localtime
|
|
import os, binascii, traceback
|
|
from unicodedata import east_asian_width
|
|
|
|
|
|
LOG_NAME = f'./logs/ssh-honeypot-{strftime("%Y%m%d%H%M%S", localtime())}.log'
|
|
|
|
LOG_COLORS = {
|
|
'red': '\033[31m',
|
|
'green': '\033[32m',
|
|
'yellow': '\033[33m',
|
|
'blue': '\033[34m',
|
|
'purple': '\033[35m',
|
|
'cyan': '\033[36m',
|
|
'white': '\033[37m',
|
|
'RESET': '\033[0m'
|
|
|
|
}
|
|
LOG_COLOR_MAP = {
|
|
'!': LOG_COLORS['yellow'], # 黄色 - 警告
|
|
'*': '', # 默认 - 普通信息
|
|
'+': LOG_COLORS['blue'], # 蓝色 - 成功
|
|
'-': '', # 默认 - 失败
|
|
'x': LOG_COLORS['red'], # 红色 - 错误
|
|
'@': '', # 默认 - 作者信息
|
|
'#': '', # 默认 - 代码信息
|
|
'RESET': LOG_COLORS['RESET'], # 重置颜色
|
|
}
|
|
|
|
# [!] 警告 warning 黄色
|
|
# [*] 普通 info
|
|
# [+] 加入成功 join success 蓝色
|
|
# [-] 加入失败 join fail
|
|
# [x] 错误 红色
|
|
# [@] 作者相关信息
|
|
# [#] 代码相关信息
|
|
|
|
LOG_DIR = "logs"
|
|
|
|
def log_init():
|
|
"""初始化日志文件"""
|
|
global LOG_NAME
|
|
os.makedirs(LOG_DIR, exist_ok=True)
|
|
|
|
|
|
def dispose_string(string):
|
|
"""处理字符串格式，替换换行符并编码"""
|
|
return string.replace('\n', '\r\n').encode('utf-8')
|
|
|
|
|
|
def get_log_color(log_type):
|
|
try:
|
|
return LOG_COLOR_MAP[log_type]
|
|
except KeyError:
|
|
return LOG_COLOR_MAP['RESET']
|
|
|
|
|
|
def custom_strip(s):
|
|
chars_to_remove = [
|
|
'\u0020', # 普通空格 (Space)
|
|
'\u3000', # 全角空格 (Ideographic Space)
|
|
'\u0003', # U+0003 (End of Text, ETX)
|
|
'\u0009', # 水平制表符 (Tab, \t)
|
|
'\u000A', # 换行符 (Line Feed, \n)
|
|
'\u000D', # 回车符 (Carriage Return, \r)
|
|
'\u000B', # 垂直制表符 (Vertical Tab, \v)
|
|
'\u000C', # 换页符 (Form Feed, \f)
|
|
]
|
|
return s.strip(''.join(chars_to_remove))
|
|
|
|
def string_width(s):
|
|
"""获取字符串的显示宽度"""
|
|
if len(s) == 0:
|
|
return 0
|
|
result = 0
|
|
for i in s:
|
|
if east_asian_width(i) in 'FWA':
|
|
result += 2
|
|
else:
|
|
result += 1
|
|
return result
|
|
|
|
|
|
def split_bytes(s, max_bytes, encoding='utf-8'):
|
|
"""将字符串按字节数分割，确保每个分片不超过 max_bytes"""
|
|
|
|
result = []
|
|
current = ""
|
|
current_bytes = 0
|
|
|
|
for char in s:
|
|
char = str(char)
|
|
char_bytes = len(char.encode(encoding))
|
|
|
|
# 如果单个字符就超过 max_bytes，直接报错
|
|
if char_bytes > max_bytes:
|
|
return -1
|
|
|
|
# 如果加上当前字符会超限，就先保存当前结果，并重置
|
|
if current_bytes + char_bytes > max_bytes:
|
|
result.append(current)
|
|
current = char
|
|
current_bytes = char_bytes
|
|
else:
|
|
current += char
|
|
current_bytes += char_bytes
|
|
|
|
# 添加最后剩余的部分
|
|
if current:
|
|
result.append(current)
|
|
return result
|
|
|
|
def send_log(log_type, log_message, color = "auto"):
|
|
"""发送日志到文件和控制台"""
|
|
|
|
log_message = custom_strip(str(log_message))
|
|
|
|
log_time = f"[{strftime('%Y-%m-%d %H:%M:%S', localtime())}]"
|
|
log = ""
|
|
|
|
# 处理颜色
|
|
color = color.lower()
|
|
if color == "auto":
|
|
color = get_log_color(log_type)
|
|
else:
|
|
if color in LOG_COLORS:
|
|
color = LOG_COLORS[color]
|
|
else:
|
|
color = LOG_COLORS['RESET']
|
|
|
|
if "\n" in log_message:
|
|
log_messages = log_message.split("\n")
|
|
for i in range(len(log_messages)):
|
|
if i == 0:
|
|
log += f"{log_time} [{log_type}] {log_messages[i]}"
|
|
else:
|
|
log += f"{string_width(log_time) * ' '} [{log_type}] {log_messages[i]}"
|
|
log += '\n'
|
|
|
|
else:
|
|
log += f"{log_time} [{log_type}] {log_message}"
|
|
# 如果目录不存在则创建
|
|
|
|
if not os.path.exists(LOG_DIR):
|
|
log_init()
|
|
|
|
with open(LOG_NAME, 'a') as f:
|
|
f.write(log + '\n')
|
|
|
|
print(f"{color}{log}{LOG_COLOR_MAP['RESET']}")
|
|
|
|
|
|
def traceback_info():
|
|
"""获取异常信息"""
|
|
tracebackInfo = traceback.format_exc()
|
|
return tracebackInfo
|
|
|
|
def is_linux_executable(filepath):
|
|
"""
|
|
检测文件是否为Linux可执行文件
|
|
:param filepath: 文件路径
|
|
:return: bool - 是否是Linux可执行文件
|
|
"""
|
|
|
|
if not os.path.isfile(filepath):
|
|
return False
|
|
|
|
try:
|
|
if filepath.endswith('.bashrc') or filepath.endswith('.sh'):
|
|
return True
|
|
|
|
with open(filepath, 'rb') as f:
|
|
# 读取文件前4个字节
|
|
header = f.read(4)
|
|
if not header:
|
|
return False
|
|
|
|
# 将字节转换为十六进制表示
|
|
hex_header = binascii.hexlify(header).decode('ascii').lower()
|
|
|
|
# ELF文件魔数: 7f 45 4c 46
|
|
if hex_header.startswith('7f454c46'):
|
|
return True
|
|
|
|
# 也可以检查shebang (#!/bin/bash等)
|
|
if header.startswith(b'#!'):
|
|
# 读取第一行检查是否是有效的shebang
|
|
f.seek(0)
|
|
first_line = f.readline().decode('utf-8', errors='ignore').strip()
|
|
if first_line.startswith('#!') and ('bash' in first_line or 'sh' in first_line or
|
|
'python' in first_line or 'perl' in first_line):
|
|
return True
|
|
|
|
except (IOError, PermissionError):
|
|
return False
|
|
|
|
return False
|
|
|
|
|
|
if __name__ == '__main__':
|
|
send_log('test', 'Running successfully!')
|