本文目录导读：
我来详细说明如何实现行为预警脚本,这是一个常见的安全监控功能。
基础行为预警系统设计
核心架构
# behavior_warning.py
import time
import json
from collections import defaultdict, deque
from datetime import datetime, timedelta
import threading
import logging
class BehaviorWarningSystem:
"""行为预警系统"""
def __init__(self):
self.rules = [] # 预警规则列表
self.events = defaultdict(lambda: deque(maxlen=1000)) # 事件队列
self.alerts = [] # 预警记录
self.logger = logging.getLogger(__name__)
def add_rule(self, rule):
"""添加预警规则"""
self.rules.append(rule)
def record_event(self, user_id, event_type, metadata=None):
"""记录事件"""
event = {
'user_id': user_id,
'type': event_type,
'timestamp': datetime.now(),
'metadata': metadata or {}
}
self.events[user_id].append(event)
self._check_rules(user_id, event)
def _check_rules(self, user_id, event):
"""检查规则触发"""
for rule in self.rules:
if rule.check(self.events[user_id], event):
alert = self._create_alert(user_id, rule, event)
self.alerts.append(alert)
self._trigger_alert(alert)
def _create_alert(self, user_id, rule, event):
"""创建预警记录"""
return {
'id': len(self.alerts) + 1,
'user_id': user_id,
'rule_name': rule.name,
'severity': rule.severity,
'event': event,
'timestamp': datetime.now(),
'status': 'active'
}
def _trigger_alert(self, alert):
"""触发预警操作"""
self.logger.warning(f"预警触发: {alert['rule_name']} - 用户: {alert['user_id']}")
# 可以添加邮件通知、Webhook等
预警规则实现
class BaseRule:
"""基础规则类"""
def __init__(self, name, severity='medium', timeframe=300):
self.name = name
self.severity = severity
self.timeframe = timeframe # 时间窗口（秒）
def check(self, events, current_event):
"""检查是否触发预警"""
raise NotImplementedError
class FrequencyRule(BaseRule):
"""频率规则 - 检测事件频率"""
def __init__(self, name, max_count, timeframe, severity='medium'):
super().__init__(name, severity, timeframe)
self.max_count = max_count
def check(self, events, current_event):
# 获取时间窗口内的事件
cutoff = datetime.now() - timedelta(seconds=self.timeframe)
recent_events = [e for e in events if e['timestamp'] > cutoff]
if len(recent_events) > self.max_count:
return True
return False
class PatternRule(BaseRule):
"""模式规则 - 检测特定行为模式"""
def __init__(self, name, pattern_type, severity='high'):
super().__init__(name, severity)
self.pattern_type = pattern_type
def check(self, events, current_event):
if self.pattern_type == 'abnormal_login':
return self._check_abnormal_login(events)
elif self.pattern_type == 'data_exfiltration':
return self._check_data_exfiltration(events)
return False
def _check_abnormal_login(self, events):
"""检测异常登录模式"""
recent_events = list(events)[-10:] # 最近10个事件
if len(recent_events) < 3:
return False
# 检测地理位置异常
locations = [e.get('metadata', {}).get('location') for e in recent_events]
if len(set(locations)) >= 3:
return True
# 检测登录时间异常
night_logins = [e for e in recent_events
if e['timestamp'].hour < 6 or e['timestamp'].hour > 22]
if len(night_logins) >= 3:
return True
return False
def _check_data_exfiltration(self, events):
"""检测数据泄露模式"""
recent_events = list(events)[-20:]
if len(recent_events) < 5:
return False
# 检测大量数据下载
download_events = [e for e in recent_events
if e['type'] == 'download']
if len(download_events) > 10:
return True
# 检测异常导出操作
export_events = [e for e in recent_events
if e['type'] == 'export']
if len(export_events) >= 5 and len(set([e['metadata'].get('target')
for e in export_events])) >= 3:
return True
return False
class ThresholdRule(BaseRule):
"""阈值规则 - 检测数值是否超过阈值"""
def __init__(self, name, field, threshold, operator='>', severity='medium'):
super().__init__(name, severity)
self.field = field
self.threshold = threshold
self.operator = operator
def check(self, events, current_event):
value = current_event.get('metadata', {}).get(self.field)
if value is None:
return False
if self.operator == '>' and value > self.threshold:
return True
elif self.operator == '<' and value < self.threshold:
return True
elif self.operator == '==' and value == self.threshold:
return True
return False
实时监控实现
class RealTimeMonitor:
"""实时监控系统"""
def __init__(self):
self.warning_system = BehaviorWarningSystem()
self.monitoring_threads = {}
self.running = False
def start_monitoring(self):
"""启动监控"""
self.running = True
monitor_thread = threading.Thread(target=self._monitor_loop)
monitor_thread.daemon = True
monitor_thread.start()
def stop_monitoring(self):
"""停止监控"""
self.running = False
def _monitor_loop(self):
"""监控主循环"""
while self.running:
# 清理过期预警
self._cleanup_old_alerts()
# 检查预警统计
self._generate_alert_statistics()
time.sleep(60) # 每分钟检查一次
def _cleanup_old_alerts(self, max_age_hours=24):
"""清理过期预警"""
cutoff = datetime.now() - timedelta(hours=max_age_hours)
self.warning_system.alerts = [
alert for alert in self.warning_system.alerts
if alert['timestamp'] > cutoff
]
def _generate_alert_statistics(self):
"""生成预警统计"""
stats = defaultdict(int)
for alert in self.warning_system.alerts:
stats[alert['rule_name']] += 1
stats[alert['severity']] += 1
# 记录统计日志
self._log_statistics(stats)
def _log_statistics(self, stats):
"""记录统计信息"""
logging.info(f"预警统计: {dict(stats)}")
def register_event_source(self, source_name, source_func):
"""注册事件源"""
def event_collector():
while self.running:
try:
events = source_func()
for event in events:
self.warning_system.record_event(
event['user_id'],
event['type'],
event.get('metadata')
)
except Exception as e:
logging.error(f"事件收集错误: {e}")
time.sleep(5)
thread = threading.Thread(target=event_collector)
thread.daemon = True
thread.start()
self.monitoring_threads[source_name] = thread
Web界面与API
from flask import Flask, request, jsonify
from flask_cors import CORS
app = Flask(__name__)
CORS(app)
monitor = RealTimeMonitor()
@app.route('/api/events', methods=['POST'])
def record_event():
"""记录事件API"""
data = request.json
monitor.warning_system.record_event(
data['user_id'],
data['type'],
data.get('metadata')
)
return jsonify({'status': 'success'})
@app.route('/api/alerts', methods=['GET'])
def get_alerts():
"""获取预警列表"""
user_id = request.args.get('user_id')
severity = request.args.get('severity')
status = request.args.get('status')
alerts = monitor.warning_system.alerts
if user_id:
alerts = [a for a in alerts if a['user_id'] == user_id]
if severity:
alerts = [a for a in alerts if a['severity'] == severity]
if status:
alerts = [a for a in alerts if a['status'] == status]
return jsonify({
'total': len(alerts),
'alerts': alerts[-100:] # 返回最近100条
})
@app.route('/api/rules', methods=['POST'])
def add_rule():
"""添加规则API"""
rule_data = request.json
rule_type = rule_data.get('type')
if rule_type == 'frequency':
rule = FrequencyRule(
rule_data['name'],
rule_data['max_count'],
rule_data['timeframe'],
rule_data.get('severity', 'medium')
)
elif rule_type == 'pattern':
rule = PatternRule(
rule_data['name'],
rule_data['pattern_type'],
rule_data.get('severity', 'high')
)
elif rule_type == 'threshold':
rule = ThresholdRule(
rule_data['name'],
rule_data['field'],
rule_data['threshold'],
rule_data.get('operator', '>'),
rule_data.get('severity', 'medium')
)
else:
return jsonify({'error': 'Unknown rule type'}), 400
monitor.warning_system.add_rule(rule)
return jsonify({'status': 'success'})
使用示例
# 使用示例
if __name__ == '__main__':
# 配置日志
logging.basicConfig(level=logging.INFO)
# 创建监控系统
monitor = RealTimeMonitor()
# 添加规则
# 1. 登录频率限制
login_rule = FrequencyRule(
'频繁登录',
max_count=10,
timeframe=300, # 5分钟内
severity='high'
)
monitor.warning_system.add_rule(login_rule)
# 2. 异常模式检测
pattern_rule = PatternRule(
'异常登录模式',
pattern_type='abnormal_login',
severity='critical'
)
monitor.warning_system.add_rule(pattern_rule)
# 3. 阈值检测
threshold_rule = ThresholdRule(
'大数据下载',
field='download_size',
threshold=1000, # MB
operator='>',
severity='medium'
)
monitor.warning_system.add_rule(threshold_rule)
# 模拟事件
import random
for i in range(100):
user_id = f'user_{random.randint(1, 5)}'
event_type = random.choice(['login', 'download', 'export', 'view'])
metadata = {
'location': random.choice(['北京', '上海', '纽约', '伦敦']),
'ip': f'192.168.1.{random.randint(1, 255)}',
'download_size': random.randint(100, 5000)
}
monitor.warning_system.record_event(user_id, event_type, metadata)
time.sleep(0.1)
# 查看预警
print("当前预警数量:", len(monitor.warning_system.alerts))
for alert in monitor.warning_system.alerts[-5:]:
print(f"预警: {alert['rule_name']} - {alert['user_id']} - {alert['severity']}")
这个行为预警系统提供了完整的框架，包括规则引擎、实时监控、事件处理和Web API,你可以根据具体需求进行扩展和定制。
