Fanjun Kong
265 lines
9.0 KiB
Python
265 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
RPM 安全扫描结果解析与报告生成
|
|
"""
|
|
import sqlite3
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from collections import defaultdict
|
|
|
|
class ScanReporter:
|
|
def __init__(self, db_path):
|
|
self.db = sqlite3.connect(db_path)
|
|
self.db.row_factory = sqlite3.Row
|
|
|
|
def parse_checksec_results(self, results_dir):
|
|
"""解析 checksec JSON 结果并入库"""
|
|
results_dir = Path(results_dir)
|
|
|
|
for json_file in results_dir.glob("*.json"):
|
|
try:
|
|
with open(json_file) as f:
|
|
data = json.load(f)
|
|
|
|
pkg_name = json_file.stem
|
|
|
|
# 获取 package_id
|
|
cur = self.db.execute(
|
|
"SELECT id FROM packages WHERE name=? AND status='success'",
|
|
(pkg_name.split('-')[0],)
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
continue
|
|
|
|
pkg_id = row[0]
|
|
|
|
# 解析每个文件的结果
|
|
for file_path, checks in data.items():
|
|
# 插入 binary 记录
|
|
cur = self.db.execute(
|
|
"INSERT INTO binaries (package_id, file_path, file_type) VALUES (?, ?, ?) RETURNING id",
|
|
(pkg_id, file_path, checks.get('type', 'unknown'))
|
|
)
|
|
binary_id = cur.fetchone()[0]
|
|
|
|
# 插入安全检查结果
|
|
self.db.execute(
|
|
"""INSERT INTO security_checks
|
|
(binary_id, pie, nx, canary, fortify, relro, bind_now)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
binary_id,
|
|
checks.get('pie', 'unknown'),
|
|
checks.get('nx', 'unknown'),
|
|
checks.get('canary', 'unknown'),
|
|
checks.get('fortify', 'unknown'),
|
|
checks.get('relro', 'unknown'),
|
|
checks.get('bind_now', 'unknown')
|
|
)
|
|
)
|
|
|
|
self.db.commit()
|
|
except Exception as e:
|
|
print(f"解析失败 {json_file}: {e}", file=sys.stderr)
|
|
|
|
def generate_statistics(self):
|
|
"""生成统计报告"""
|
|
stats = {}
|
|
|
|
# 总体统计
|
|
cur = self.db.execute("SELECT COUNT(*) FROM packages WHERE status='success'")
|
|
stats['total_packages'] = cur.fetchone()[0]
|
|
|
|
cur = self.db.execute("SELECT COUNT(*) FROM binaries")
|
|
stats['total_binaries'] = cur.fetchone()[0]
|
|
|
|
# 定义每个特性的启用判断规则
|
|
feature_rules = {
|
|
'pie': {
|
|
'enabled': ['PIE Enabled'],
|
|
'disabled': ['PIE Disabled'],
|
|
'label': 'PIE'
|
|
},
|
|
'nx': {
|
|
'enabled': ['NX enabled', 'NX enabled'],
|
|
'disabled': ['NX disabled', 'NX disabled'],
|
|
'label': 'NX'
|
|
},
|
|
'canary': {
|
|
'enabled': ['Canary Found'],
|
|
'disabled': ['No Canary Found', 'No canary found'],
|
|
'label': 'CANARY'
|
|
},
|
|
'fortify': {
|
|
'enabled_values': lambda x: x and x != '0' and x != 'unknown',
|
|
'label': 'FORTIFY'
|
|
},
|
|
'relro': {
|
|
'enabled': ['Full RELRO'],
|
|
'partial': ['Partial RELRO'],
|
|
'disabled': ['No RELRO'],
|
|
'label': 'RELRO'
|
|
}
|
|
}
|
|
|
|
stats['coverage'] = {}
|
|
|
|
for feature, rules in feature_rules.items():
|
|
cur = self.db.execute(f"""
|
|
SELECT
|
|
{feature},
|
|
COUNT(*) as count
|
|
FROM security_checks
|
|
GROUP BY {feature}
|
|
""")
|
|
|
|
feature_stats = {}
|
|
total = 0
|
|
enabled = 0
|
|
|
|
for row in cur:
|
|
value = row[0]
|
|
count = row[1]
|
|
feature_stats[value] = count
|
|
total += count
|
|
|
|
# 根据规则判断是否启用
|
|
if 'enabled' in rules and value in rules['enabled']:
|
|
enabled += count
|
|
elif 'enabled_values' in rules and rules['enabled_values'](value):
|
|
enabled += count
|
|
|
|
coverage_rate = f"{enabled/total*100:.2f}%" if total > 0 else "0.00%"
|
|
|
|
stats['coverage'][feature] = {
|
|
'label': rules['label'],
|
|
'details': feature_stats,
|
|
'rate': coverage_rate,
|
|
'total': total,
|
|
'enabled': enabled
|
|
}
|
|
|
|
return stats
|
|
|
|
def export_csv(self, output_file):
|
|
"""导出 CSV 报告"""
|
|
cur = self.db.execute("""
|
|
SELECT
|
|
p.name,
|
|
b.file_path,
|
|
s.pie, s.nx, s.canary, s.fortify, s.relro, s.bind_now
|
|
FROM packages p
|
|
JOIN binaries b ON p.id = b.package_id
|
|
JOIN security_checks s ON b.id = s.binary_id
|
|
ORDER BY p.name, b.file_path
|
|
""")
|
|
|
|
with open(output_file, 'w') as f:
|
|
f.write("Package,File,PIE,NX,Canary,Fortify,RELRO,BIND_NOW\n")
|
|
for row in cur:
|
|
f.write(','.join(str(x) for x in row) + '\n')
|
|
|
|
def print_report(self):
|
|
"""打印统计报告"""
|
|
stats = self.generate_statistics()
|
|
|
|
# Markdown 格式输出
|
|
print("# RPM 安全扫描统计报告")
|
|
print()
|
|
print(f"**扫描包数量**: {stats['total_packages']}")
|
|
print(f"**二进制文件数**: {stats['total_binaries']}")
|
|
print()
|
|
print("## 安全特性覆盖率")
|
|
print()
|
|
|
|
# 按顺序输出: PIE, NX, CANARY, FORTIFY, RELRO
|
|
feature_order = ['pie', 'nx', 'canary', 'fortify', 'relro']
|
|
|
|
for feature in feature_order:
|
|
if feature not in stats['coverage']:
|
|
continue
|
|
|
|
data = stats['coverage'][feature]
|
|
label = data['label']
|
|
rate = data['rate']
|
|
details = data['details']
|
|
|
|
# 输出标题和覆盖率
|
|
print(f"### {label}")
|
|
print()
|
|
print(f"**覆盖率**: {rate}")
|
|
print()
|
|
print("| 状态 | 数量 | 百分比 |")
|
|
print("|------|------|--------|")
|
|
|
|
# 输出详细统计（按数量降序排列）
|
|
sorted_details = sorted(details.items(), key=lambda x: x[1], reverse=True)
|
|
for value, count in sorted_details:
|
|
percentage = count * 100 / data['total'] if data['total'] > 0 else 0
|
|
print(f"| {value} | {count} | {percentage:.2f}% |")
|
|
|
|
print()
|
|
|
|
print("## 综合安全评分")
|
|
print()
|
|
|
|
# 计算综合安全评分
|
|
total_binaries = stats['total_binaries']
|
|
if total_binaries > 0:
|
|
pie_enabled = stats['coverage']['pie']['enabled']
|
|
nx_enabled = stats['coverage']['nx']['enabled']
|
|
canary_enabled = stats['coverage']['canary']['enabled']
|
|
relro_full = stats['coverage']['relro']['enabled']
|
|
|
|
# 计算所有保护都开启的文件数（取最小值）
|
|
fully_protected = min(pie_enabled, nx_enabled, canary_enabled, relro_full)
|
|
security_score = fully_protected * 100 / total_binaries
|
|
|
|
print(f"- **综合安全评分**: {security_score:.2f}%")
|
|
print(f"- **完全加固的文件**: {fully_protected}/{total_binaries}")
|
|
print(f"- **标准**: PIE + NX + CANARY + Full RELRO")
|
|
print()
|
|
|
|
def main():
|
|
# 设置默认数据库路径
|
|
default_db = "scan_results.db"
|
|
|
|
# 解析参数
|
|
db_path = default_db
|
|
args_start = 0
|
|
|
|
# 如果第一个参数不是选项，则作为数据库路径
|
|
if len(sys.argv) > 1 and not sys.argv[1].startswith('--'):
|
|
db_path = sys.argv[1]
|
|
args_start = 1
|
|
|
|
reporter = ScanReporter(db_path)
|
|
|
|
# 处理选项参数
|
|
if '--parse-results' in sys.argv:
|
|
idx = sys.argv.index('--parse-results')
|
|
# 检查参数是否越界
|
|
if idx + 1 >= len(sys.argv):
|
|
print("错误: --parse-results 需要指定结果目录路径", file=sys.stderr)
|
|
sys.exit(1)
|
|
results_dir = sys.argv[idx + 1]
|
|
print(f"解析结果目录: {results_dir}")
|
|
reporter.parse_checksec_results(results_dir)
|
|
|
|
if '--export-csv' in sys.argv:
|
|
idx = sys.argv.index('--export-csv')
|
|
# 检查参数是否越界
|
|
if idx + 1 >= len(sys.argv):
|
|
print("错误: --export-csv 需要指定输出 CSV 文件路径", file=sys.stderr)
|
|
sys.exit(1)
|
|
output_file = sys.argv[idx + 1]
|
|
print(f"导出 CSV: {output_file}")
|
|
reporter.export_csv(output_file)
|
|
|
|
reporter.print_report()
|
|
|
|
if __name__ == '__main__':
|
|
main()
|