def connect_all(self):
"""建立所有数据库连接"""
connections =
for cfg in self.configs:
try:
conn = pymysql.connect(cfg)
connections.append(conn)
logging.info(f"成功连接数据库：{cfg'host'}:{cfg'port'}")
except Exception as e:
logging.error(f"连接失败 {cfg'host'}：{str(e)}")
return connections
====== 增强型数据采集 ======
def enhanced_collect(conn) -> Dict:
"""增强型数据采集（包含密码字段检测）"""
data = {}
with conn.cursor() as cursor:
获取所有数据库
cursor.execute("SHOW DATABASES")
databases = db0 for db in cursor.fetchall()
for db in databases:
if db in ('mysql', 'sys'): continue
datadb = {}
try:
cursor.execute(f"USE
cursor.execute("SHOW TABLES")
tables = tbl0 for tbl in cursor.fetchall()
for tbl in tables:
获取表结构
cursor.execute(f"DESCRIBE
columns = col0 for col in cursor.fetchall()
检测敏感字段
sensitive_cols =
for col in columns:
if re.search(r'passw(or)?dpwdhashsalt', col, re.I):
sensitive_cols.append(col)
获取样本并分析加密
crypto_analysis = {}
if sensitive_cols:
cursor.execute(f"SELECT {','.join(sensitive_cols)} FROM
samples = cursor.fetchall()
for idx, col in enumerate(sensitive_cols):
sample = samples0idx if samples else ''
crypto_types = CryptoDetector.detect_crypto_type(sample)
crypto_analysiscol = {
'types': crypto_types,
'advice': CryptoDetector.get_decrypt_advice(crypto_types)
}
datadbtbl = {
'columns': columns,
'sensitive': crypto_analysis
}
except Exception as e:
logging.error(f"采集失败 {db}.{tbl}：{str(e)}")
return data
====== 改进后的主流程 ======
def main():
初始化多数据库连接
db_mgr = DBConnectionManager("multi_db_config.json")
connections = db_mgr.connect_all()
all_reports =
for conn in connections:
try:
执行增强采集
report = enhanced_collect(conn)
all_reports.append(report)
生成渗透建议报告
analysis_prompt = generate_analysis_prompt(report)
ai_advice = openai.ChatCompletion.create(
model="gpt-4",
messages={"role": "user", "content": analysis_prompt}
)
保存结果到Markdown
save_markdown_report(report, ai_advice)
finally:
conn.close()
合并所有报告
merge_reports(all_reports)
核心改进说明：
1. 多数据库支持
json
// multi_db_config.json
{
"host": "db1.example.com",
"user": "audit_user",
"password": "gAAAAABk...（加密凭证）",
"encrypted": true,
"port": 3306
},
{
"host": "db2.example.com",
"user": "readonly_user",
"password": "plaintext_pass",
"port": 3307
}
2. 加密类型识别逻辑
python
输入样本检测示例
sample = "5f4dcc3b5aa765d61d8327deb882cf99" MD5
CryptoDetector.detect_crypto_type(sample)
输出: {'md5': 0.9}
解密建议生成
advice = CryptoDetector.get_decrypt_advice({'md5': 0.9})
print(advice)
输出: "MD5哈希：使用彩虹表碰撞（如hashcat -m 0）或在线解密网站"
3. 渗透建议报告示例
markdown
安全审计报告 - 192.168.1.100
敏感字段分析
数据库 表 字段 加密类型 渗透建议
user_db accounts password bcrypt 需GPU集群暴力破解（约$1/10亿次）
order_db transactions card_token AES 查找配置文件中的AES密钥
AI建议
尝试以下渗透路径：
1. 从web应用的/config目录查找aes_key.txt
2. 使用JohnTheRipper进行bcrypt爆破
3. 检查数据库备份文件中的测试账号
"""建立所有数据库连接"""
connections =
for cfg in self.configs:
try:
conn = pymysql.connect(cfg)
connections.append(conn)
logging.info(f"成功连接数据库：{cfg'host'}:{cfg'port'}")
except Exception as e:
logging.error(f"连接失败 {cfg'host'}：{str(e)}")
return connections
====== 增强型数据采集 ======
def enhanced_collect(conn) -> Dict:
"""增强型数据采集（包含密码字段检测）"""
data = {}
with conn.cursor() as cursor:
获取所有数据库
cursor.execute("SHOW DATABASES")
databases = db0 for db in cursor.fetchall()
for db in databases:
if db in ('mysql', 'sys'): continue
datadb = {}
try:
cursor.execute(f"USE
{db}")cursor.execute("SHOW TABLES")
tables = tbl0 for tbl in cursor.fetchall()
for tbl in tables:
获取表结构
cursor.execute(f"DESCRIBE
{tbl}")columns = col0 for col in cursor.fetchall()
检测敏感字段
sensitive_cols =
for col in columns:
if re.search(r'passw(or)?dpwdhashsalt', col, re.I):
sensitive_cols.append(col)
获取样本并分析加密
crypto_analysis = {}
if sensitive_cols:
cursor.execute(f"SELECT {','.join(sensitive_cols)} FROM
{tbl} LIMIT 5")samples = cursor.fetchall()
for idx, col in enumerate(sensitive_cols):
sample = samples0idx if samples else ''
crypto_types = CryptoDetector.detect_crypto_type(sample)
crypto_analysiscol = {
'types': crypto_types,
'advice': CryptoDetector.get_decrypt_advice(crypto_types)
}
datadbtbl = {
'columns': columns,
'sensitive': crypto_analysis
}
except Exception as e:
logging.error(f"采集失败 {db}.{tbl}：{str(e)}")
return data
====== 改进后的主流程 ======
def main():
初始化多数据库连接
db_mgr = DBConnectionManager("multi_db_config.json")
connections = db_mgr.connect_all()
all_reports =
for conn in connections:
try:
执行增强采集
report = enhanced_collect(conn)
all_reports.append(report)
生成渗透建议报告
analysis_prompt = generate_analysis_prompt(report)
ai_advice = openai.ChatCompletion.create(
model="gpt-4",
messages={"role": "user", "content": analysis_prompt}
)
保存结果到Markdown
save_markdown_report(report, ai_advice)
finally:
conn.close()
合并所有报告
merge_reports(all_reports)
核心改进说明：
1. 多数据库支持
json
// multi_db_config.json
{
"host": "db1.example.com",
"user": "audit_user",
"password": "gAAAAABk...（加密凭证）",
"encrypted": true,
"port": 3306
},
{
"host": "db2.example.com",
"user": "readonly_user",
"password": "plaintext_pass",
"port": 3307
}
2. 加密类型识别逻辑
python
输入样本检测示例
sample = "5f4dcc3b5aa765d61d8327deb882cf99" MD5
CryptoDetector.detect_crypto_type(sample)
输出: {'md5': 0.9}
解密建议生成
advice = CryptoDetector.get_decrypt_advice({'md5': 0.9})
print(advice)
输出: "MD5哈希：使用彩虹表碰撞（如hashcat -m 0）或在线解密网站"
3. 渗透建议报告示例
markdown
安全审计报告 - 192.168.1.100
敏感字段分析
数据库 表 字段 加密类型 渗透建议
user_db accounts password bcrypt 需GPU集群暴力破解（约$1/10亿次）
order_db transactions card_token AES 查找配置文件中的AES密钥
AI建议
尝试以下渗透路径：
1. 从web应用的/config目录查找aes_key.txt
2. 使用JohnTheRipper进行bcrypt爆破
3. 检查数据库备份文件中的测试账号
