怎样实现行为预警脚本

wen 实用脚本 2026-06-29 196

本文目录导读：

怎样实现行为预警脚本

基础行为预警系统设计
预警规则实现
实时监控实现
Web界面与API
使用示例

我来详细说明如何实现行为预警脚本,这是一个常见的安全监控功能。

基础行为预警系统设计

核心架构

# 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,你可以根据具体需求进行扩展和定制。

本文地址： https://www.bmjup.com/post/11833.html