安全审计实战指南：Python 日志实现与 GDPR 合规深度解析

安全审计实战指南：Python 日志实现与 GDPR 合规深度解析

1. 技术分析

1.1 安全审计概述

安全审计是对系统安全状态的评估和检查：

审计类型
合规审计: 符合法规要求
安全审计: 评估安全状态
性能审计: 评估系统性能
财务审计: 评估财务安全

审计目的:
发现漏洞
验证合规
评估风险
改进安全

1.2 合规标准

合规标准
GDPR: 欧盟数据保护条例
HIPAA: 医疗信息保护
PCI DSS: 支付卡行业安全标准
ISO 27001: 信息安全管理

合规要求:
数据保护
隐私保护
安全控制
审计追踪

1.3 审计流程

审计流程
计划阶段: 确定范围和目标
执行阶段: 收集证据
报告阶段: 提交审计报告
整改阶段: 修复问题

审计方法:
问卷调查
技术检测
文档审查
现场检查

2. 核心功能实现

2.1 审计日志系统

import logging
from datetime import datetime
import json

class AuditLogger:
def __init__(self, log_file='audit.log'):
self.logger = logging.getLogger('audit')
self.logger.setLevel(logging.INFO)

handler = logging.FileHandler(log_file)
formatter = logging.Formatter('%(asctime)s – %(levelname)s – %(message)s')
handler.setFormatter(formatter)

self.logger.addHandler(handler)

self.logs = []

def log_event(self, event_type, user, resource, action, details=None):
event = {
'timestamp': datetime.now().isoformat(),
'event_type': event_type,
'user': user,
'resource': resource,
'action': action,
'details': details or {}
}

self.logs.append(event)

log_message = json.dumps(event)
self.logger.info(log_message)

def get_logs_by_user(self, user):
return [log for log in self.logs if log['user'] == user]

def get_logs_by_type(self, event_type):
return [log for log in self.logs if log['event_type'] == event_type]

def search_logs(self, **filters):
results = self.logs

for key, value in filters.items():
results = [log for log in results if log.get(key) == value]

return results

def generate_report(self, start_time=None, end_time=None):
filtered = self.logs

if start_time:
filtered = [log for log in filtered if log['timestamp'] >= start_time]

if end_time:
filtered = [log for log in filtered if log['timestamp'] <= end_time]

report = {
'total_events': len(filtered),
'event_types': {},
'top_users': {},
'top_resources': {}
}

for log in filtered:
et = log['event_type']
report['event_types'][et] = report['event_types'].get(et, 0) + 1

user = log['user']
report['top_users'][user] = report['top_users'].get(user, 0) + 1

resource = log['resource']
report['top_resources'][resource] = report['top_resources'].get(resource, 0) + 1

return report

2.2 合规检查器

class ComplianceChecker:
def __init__(self):
self.standards = {}

def add_standard(self, name, requirements):
self.standards[name] = requirements

def check_compliance(self, system_info):
results = {}

for standard, requirements in self.standards.items():
results[standard] = {
'compliant': True,
'issues': []
}

for requirement in requirements:
check = requirement['check']
result = check(system_info)

if not result:
results[standard]['compliant'] = False
results[standard]['issues'].append(requirement['description'])

return results

def generate_compliance_report(self, system_info):
results = self.check_compliance(system_info)

report = {
'overall_compliance': all(s['compliant'] for s in results.values()),
'standards': results,
'summary': {
'compliant': sum(1 for s in results.values() if s['compliant']),
'total': len(results)
}
}

return report

2.3 漏洞扫描器

import requests

class VulnerabilityScanner:
def __init__(self):
self.vulnerabilities = []

def scan_web_app(self, url):
results = []

checks = [
self._check_https,
self._check_security_headers,
self._check_xss_vulnerability,
self._check_sql_injection
]

for check in checks:
try:
result = check(url)
results.extend(result)
except Exception as e:
results.append({
'type': 'error',
'message': f"Error during {check.__name__}: {str(e)}"
})

self.vulnerabilities.extend(results)
return results

def _check_https(self, url):
results = []

if not url.startswith('https://'):
results.append({
'severity': 'high',
'issue': 'HTTPS not used',
'description': 'The application is not using HTTPS'
})

return results

def _check_security_headers(self, url):
results = []

try:
response = requests.get(url)
headers = response.headers

security_headers = [
('X-Content-Type-Options', 'nosniff'),
('X-Frame-Options', 'DENY'),
('X-XSS-Protection', '1; mode=block'),
('Content-Security-Policy', None)
]

for header, expected in security_headers:
if header not in headers:
results.append({
'severity': 'medium',
'issue': f"Missing {header}",
'description': f"The {header} security header is missing"
})
elif expected and headers[header] != expected:
results.append({
'severity': 'low',
'issue': f"{header} not properly configured",
'description': f"Expected '{expected}', got '{headers[header]}'"
})
except Exception as e:
results.append({
'severity': 'info',
'issue': 'Could not check headers',
'description': str(e)
})

return results

def _check_xss_vulnerability(self, url):
results = []

test_payload = '<script>alert(1)</script>'

try:
response = requests.get(url, params={'test': test_payload})

if test_payload in response.text:
results.append({
'severity': 'high',
'issue': 'XSS vulnerability detected',
'description': 'The application is vulnerable to cross-site scripting'
})
except Exception as e:
results.append({
'severity': 'info',
'issue': 'Could not test XSS',
'description': str(e)
})

return results

def _check_sql_injection(self, url):
results = []

test_payload = "' OR '1'='1"

try:
response = requests.get(url, params={'id': test_payload})

if 'error' in response.text.lower() or 'mysql' in response.text.lower():
results.append({
'severity': 'high',
'issue': 'SQL injection vulnerability detected',
'description': 'The application may be vulnerable to SQL injection'
})
except Exception as e:
results.append({
'severity': 'info',
'issue': 'Could not test SQL injection',
'description': str(e)
})

return results

def generate_vulnerability_report(self):
report = {
'total_vulnerabilities': len(self.vulnerabilities),
'by_severity': {
'critical': 0,
'high': 0,
'medium': 0,
'low': 0,
'info': 0
},
'vulnerabilities': self.vulnerabilities
}

for vuln in self.vulnerabilities:
severity = vuln.get('severity', 'info')

if severity in report['by_severity']:
report['by_severity'][severity] += 1

return report

3. 性能对比

3.1 合规标准对比

3.2 审计工具对比

3.3 安全评分对比

4. 最佳实践

4.1 审计日志示例

def audit_logging_example():
logger = AuditLogger()

logger.log_event('login', 'user123', '/login', 'success')
logger.log_event('access', 'user123', '/api/data', 'read', {'data_id': '123'})
logger.log_event('modification', 'user123', '/api/data/123', 'update', {'field': 'name'})

report = logger.generate_report()
print(f"Audit report: {json.dumps(report, indent=2)}")

4.2 合规检查示例

def compliance_check_example():
checker = ComplianceChecker()

gdpr_requirements = [
{'check': lambda info: info.get('data_encryption') == True,
'description': 'Data encryption not enabled'},
{'check': lambda info: info.get('privacy_policy') == True,
'description': 'Privacy policy not present'},
{'check': lambda info: info.get('data_retention_policy') == True,
'description': 'Data retention policy not defined'}
]

checker.add_standard('GDPR', gdpr_requirements)

system_info = {
'data_encryption': True,
'privacy_policy': True,
'data_retention_policy': False
}

report = checker.generate_compliance_report(system_info)
print(f"Compliance report: {json.dumps(report, indent=2)}")

5. 总结

安全审计与合规是信息安全的重要组成部分：

对比数据如下：

• GDPR最严格
• Nessus功能最全面
• CVSS评分最常用
• 推荐定期进行安全审计

安全审计与合规需要定期进行，确保系统持续符合安全标准。