You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
133 lines
5.3 KiB
133 lines
5.3 KiB
#mysql
|
|
#pip install mysql-connector-python
|
|
import pexpect
|
|
import shlex
|
|
import re
|
|
import mysql.connector
|
|
from mysql.connector import Error
|
|
from tools.ToolBase import ToolBase
|
|
|
|
class MysqlTool(ToolBase):
|
|
|
|
def test_empty_password_mysql_connection(self,host, username='root'):
|
|
"""
|
|
测试使用空密码连接到指定 MySQL 服务器。
|
|
|
|
参数:
|
|
host (str): MySQL 服务器的主机地址，例如 'haitutech.cn'
|
|
username (str): MySQL 用户名，默认值为 'root'
|
|
"""
|
|
try:
|
|
# 尝试使用空密码连接 MySQL
|
|
connection = mysql.connector.connect(
|
|
host=host, # 主机地址
|
|
user=username, # 用户名
|
|
password='', # 空密码
|
|
connection_timeout=10 # 设置10秒连接超时
|
|
)
|
|
if connection.is_connected():
|
|
res = f"成功连接到 {host}，用户 {username} 使用空密码"
|
|
connection.close() # 关闭连接以释放资源
|
|
except Error as e:
|
|
# 捕获并打印连接错误
|
|
res = f"连接失败: {host} - {e}"
|
|
return res
|
|
|
|
def extract_mysql_password(self,command_line: str) -> str:
|
|
"""
|
|
从 MySQL 命令行中提取 -p 参数后的密码值
|
|
|
|
支持场景：
|
|
-p'' → 返回空字符串
|
|
-p'password' → 返回 'password'
|
|
-p"password" → 返回 'password'
|
|
-ppassword → 返回 'password'
|
|
-p password → 返回 'password'
|
|
"""
|
|
# 正则匹配模式
|
|
pattern = r'''
|
|
-p # 匹配 -p 参数
|
|
(?: # 非捕获分组
|
|
\s* # 允许空格（如 -p password）
|
|
(["']?) # 捕获引号类型（单引号/双引号/无引号）
|
|
(.*?) # 捕获密码内容
|
|
\1 # 闭合引号（与开头引号一致）
|
|
| # 或
|
|
(\S+) # 直接捕获无空格的密码（如 -ppassword）
|
|
)
|
|
'''
|
|
matches = re.search(pattern, command_line, re.VERBOSE)
|
|
|
|
if not matches:
|
|
return ""
|
|
|
|
# 提取密码（处理三种情况：引号包裹、无引号、直接拼接）
|
|
password = matches.group(2) or matches.group(3) or ""
|
|
return password
|
|
|
|
def validate_instruction(self, instruction):
|
|
timeout = 60
|
|
#modified_code = "mysql空密码登录测试"
|
|
instr = instruction.replace("--ssl-mode=DISABLED","--ssl=0") #mariaDB 没有ssl-mode参数
|
|
# if "--ssl=0" not in instr:
|
|
# instr = instr + " --ssl=0"
|
|
return instr,timeout
|
|
|
|
def do_worker_pexpect(self,str_instruction,timeout,ext_params):
|
|
try:
|
|
#safe_command = shlex.quote(str_instruction)
|
|
#cmd = f"bash -c {safe_command}"
|
|
strpwsd = self.extract_mysql_password(str_instruction)
|
|
|
|
result = ""
|
|
exc_do = pexpect.spawn('bash',['-c',str_instruction],timeout=timeout,encoding='utf-8')#spawn 第一个参数是可执行文件
|
|
index = exc_do.expect([
|
|
pexpect.TIMEOUT,
|
|
pexpect.EOF,
|
|
'Enter password:'
|
|
])
|
|
result += str(exc_do.before)
|
|
if index == 0 or index ==1:
|
|
return result
|
|
elif index==2:#针对要输入密码的情况，暂时智能输入个空字符
|
|
result += "Enter password:\n"
|
|
exc_do.sendline(strpwsd) # 输入空密码后不知道会有多少种情况，密码不对，密码对
|
|
index = exc_do.expect([pexpect.TIMEOUT, pexpect.EOF])
|
|
result += str(exc_do.before)
|
|
else:
|
|
print("遇到其他输出！")
|
|
return result
|
|
except Exception as e:
|
|
return f"执行错误: {str(e)}"
|
|
|
|
#对于非sh命令调用的工具，自己实现命令执行的内容 --#2025-3-24暂时不使用
|
|
def execute_instruction(self, instruction_old):
|
|
ext_params = self.create_extparams()
|
|
# 第一步：验证指令合法性
|
|
instruction, timeout = self.validate_instruction(instruction_old)
|
|
if not instruction:
|
|
ext_params.is_user = True
|
|
return False, instruction_old, "该指令暂不执行！由用户确认是否要兼容支持", "", ext_params # 未
|
|
# 过滤修改后的指令是否需要判重？同样指令再执行结果一致？待定---#？
|
|
|
|
# 第二步：执行指令
|
|
output = self.do_worker_pexpect(instruction, timeout, ext_params)
|
|
|
|
# 第三步：分析执行结果
|
|
if isinstance(output, bytes): # 若是bytes则转成str
|
|
output = output.decode('utf-8', errors='ignore')
|
|
|
|
analysis = self.analyze_result(output, instruction, "", "")
|
|
|
|
if not analysis: # analysis为“” 不提交LLM
|
|
ext_params.is_user = True
|
|
return False, instruction, analysis, output, ext_params
|
|
return True, instruction, analysis, output, ext_params
|
|
|
|
def analyze_result(self, result,instruction,stderr,stdout):
|
|
#如果有--------，把这些-去除掉
|
|
if result:
|
|
result = result.replace("--------------","")
|
|
else:
|
|
result = ""
|
|
return result
|