规则引擎在安全监控中的实战应用从异常登录检测案例看中小企业如何防范内部威胁与外部攻击

---

一、先说说，为什么中小企业总被”按在地上摩擦”

我见过太多中小企业的老板，IT预算紧巴巴的，安全产品买了又买，结果出了事才发现——设备在那儿躺着，规则从没配置过。

有个做电商的朋友跟我吐槽，去年黑五期间，他们的后台被登录了三百多次，其中一百多次是从境外IP来的，但告警系统一声不吭。后来查日志才发现，攻击者用的是撞库来的账号，密码是他们用的”123456”。

这事让我意识到一个问题：安全不是买设备，而是建规则。

规则引擎，听起来很技术，但说白了就是：告诉系统”什么情况不对劲”。

---

二、什么是规则引擎？用大白话讲清楚

规则引擎（Rule Engine）说白了就是一个”决策大脑”。

你给它一堆事实（比如：谁在登录、从哪儿来、用的什么设备），它根据你事先写好的规则，判断这是正常行为还是异常行为，然后做出响应。

---

举个最直观的例子

想象你在公司门口装了个门禁：

• 普通保安：看一眼身份证，放行
• 规则引擎门禁：先看时间、再看部门、再看设备指纹、再看历史行为，综合判断才放行

规则引擎就是这么回事——用一连串”如果…就…“的逻辑，帮系统做判断。

---

三、中小企业安全监控的三大痛点

痛点一：告警太多，看不过来

很多公司用了SIEM（安全信息和事件管理）系统，结果每天收到几千条告警，运维人员看都看不过来，最后只能”全选标记为已读”。

问题在哪？规则不够精细。

---

痛点二：内外兼忧，分不清谁是敌人

外部攻击者用脚本跑账号，内部员工忘记改密码把账号共享给外包，这两种行为在日志里看起来可能差不多——都是”异地登录”。

规则引擎的价值就在这里：它能区分”异常”的类型，告诉你是外部攻击还是内部泄密。

---

痛点三：响应太慢，出了事才想起来

很多中小企业的流程是：出问题→打电话问安全厂商→厂商给方案→配置半天→再出问题。

规则引擎配合自动化响应，可以做到：检测到异常→立即执行处置动作，比如：

• 临时锁定账号
• 发送二次验证
• 记录完整日志供后续分析

---

四、异常登录检测——实战案例详解

4.1 场景设定

假设有一家做SaaS服务的中小企业，有200名员工，使用内部OA系统管理业务。

他们的系统架构：

• 身份认证：自研登录系统 + LDAP
• 日志来源：Web服务器、AD域控、 VPN网关、堡垒机
• 安全产品：基础防火墙 + 一台SIEM设备（买了三年从没配置过规则）

---

4.2 异常登录的四种典型模式

模式一：暴力破解

攻击者用账号密码字典，对一个账号反复尝试登录。

特征：

• 同一IP短时间内对多个账号尝试登录
• 同一账号短时间内多次失败后成功

---

模式二：异地登录

员工平时在北京登录，突然从境外IP登录。

特征：

• 登录地点与历史地点差异过大
• 短时间内从两个地理距离很远的地方登录

---

模式三：非工作时间登录

正常员工9点到6点登录，有人在凌晨3点登录了管理员账号。

特征：

• 登录时间偏离正常工作时间
• 访问了敏感资源

---

模式四：代理/ VPN滥用

攻击者使用境外代理IP，绕过地域限制。

特征：

• IP属于已知代理/VPN服务商
• 同一代理IP被多个账号使用

---

五、用规则引擎实现异常登录检测

下面我用Python + Drools规则引擎的方式，给大家演示一个完整的规则实现。

5.1 数据模型定义

首先，我们需要定义事件的数据结构：

from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional
from enum import Enum

class RiskLevel(Enum):
    LOW = "低"
    MEDIUM = "中"
    HIGH = "高"
    CRITICAL = "危急"

@dataclass
class LoginEvent:
    """登录事件"""
    event_id: str              # 事件唯一ID
    timestamp: datetime        # 发生时间
    username: str              # 登录账号
    source_ip: str             # 来源IP
    user_agent: str            # 用户代理
    login_result: bool         # 登录是否成功
    auth_method: str           # 认证方式（密码/OTP/证书）
    
    # 附加信息
    geolocation: Optional[str] = None   # 地理位置
    device_fingerprint: Optional[str] = None  # 设备指纹
    is_vpn: bool = False                 # 是否使用VPN
    is_proxy: bool = False               # 是否使用代理

@dataclass
class UserProfile:
    """用户行为基线"""
    username: str
    usual_locations: List[str]           # 常用登录地点
    usual_ips: List[str]                 # 常用IP段
    usual_time_range: tuple              # (开始小时, 结束小时)
    usual_devices: List[str]             # 常用设备指纹
    role: str                            # 角色（admin/user/guest）
    last_login_time: Optional[datetime] = None
    failed_attempts: int = 0             # 连续失败次数
    account_locked: bool = False         # 账号是否锁定

@dataclass
class SecurityAlert:
    """安全告警"""
    alert_id: str
    rule_name: str
    risk_level: RiskLevel
    event: LoginEvent
    description: str
    suggested_action: str
    timestamp: datetime

---

5.2 规则引擎核心实现

import hashlib
import itertools
from collections import defaultdict
from datetime import timedelta

class LoginSecurityRuleEngine:
    """登录安全规则引擎"""
    
    def __init__(self):
        # 用户行为档案库
        self.user_profiles: dict = {}
        # 事件窗口（滑动时间窗口）
        self.event_window: dict = defaultdict(list)
        # 告警输出
        self.alerts: List[SecurityAlert] = []
        # 规则版本
        self.rule_version = "1.0"
    
    def register_user_profile(self, profile: UserProfile):
        """注册用户行为基线"""
        self.user_profiles[profile.username] = profile
        print(f"[规则引擎] 已注册用户基线: {profile.username} "
              f"(角色:{profile.role}, 地点:{profile.usual_locations})")
    
    def evaluate(self, event: LoginEvent) -> List[SecurityAlert]:
        """
        对登录事件执行规则评估
        返回产生的告警列表
        """
        alerts = []
        
        # 规则1: 暴力破解检测
        brute_force_alert = self._check_brute_force(event)
        if brute_force_alert:
            alerts.append(brute_force_alert)
        
        # 规则2: 异地登录检测
        location_alert = self._check_anomalous_location(event)
        if location_alert:
            alerts.append(location_alert)
        
        # 规则3: 非工作时间登录
        time_alert = self._check_unusual_time(event)
        if time_alert:
            alerts.append(time_alert)
        
        # 规则4: 代理/VPN滥用
        proxy_alert = self._check_proxy_abuse(event)
        if proxy_alert:
            alerts.append(proxy_alert)
        
        # 规则5: 账号异常活跃
        activity_alert = self._check_account_activity(event)
        if activity_alert:
            alerts.append(activity_alert)
        
        # 更新用户档案
        self._update_user_profile(event)
        
        return alerts
    
    def _check_brute_force(self, event: LoginEvent) -> Optional[SecurityAlert]:
        """
        规则1: 暴力破解检测
        
        逻辑：
        - 如果登录失败，增加失败计数
        - 如果失败次数 >= 5次，产生高危告警
        - 如果同一IP在5分钟内尝试登录超过10个不同账号，产生危急告警
        """
        username = event.username
        window = self.event_window[username]
        
        # 清理过期事件（超过30分钟）
        cutoff = datetime.now() - timedelta(minutes=30)
        self.event_window[username] = [
            e for e in window if e.timestamp > cutoff
        ]
        window = self.event_window[username]
        
        # 添加当前事件
        window.append(event)
        
        # 统计失败次数
        recent_failures = [
            e for e in window 
            if not e.login_result and e.timestamp > datetime.now() - timedelta(minutes=10)
        ]
        
        user_profile = self.user_profiles.get(username)
        
        # 条件A: 短时间内多次失败
        if len(recent_failures) >= 5:
            return SecurityAlert(
                alert_id=self._gen_alert_id("BRUTE_FORCE"),
                rule_name="暴力破解检测-多次失败",
                risk_level=RiskLevel.HIGH,
                event=event,
                description=f"账号 {username} 在10分钟内失败登录 {len(recent_failures)} 次，疑似暴力破解",
                suggested_action="锁定账号并通知管理员，检查该IP是否在其他系统有异常行为",
                timestamp=datetime.now()
            )
        
        # 条件B: 同一IP多账号尝试（横向移动）
        ip_events = [e for e in window if e.source_ip == event.source_ip]
        unique_accounts = set(e.username for e in ip_events)
        
        if len(unique_accounts) >= 10:
            return SecurityAlert(
                alert_id=self._gen_alert_id("CREDENTIAL_STUFFING"),
                rule_name="撞库攻击检测-多账号尝试",
                risk_level=RiskLevel.CRITICAL,
                event=event,
                description=f"IP {event.source_ip} 在30分钟内尝试了 {len(unique_accounts)} 个不同账号，疑似撞库攻击",
                suggested_action="立即封禁该IP，检查所有受影响账号的密码强度",
                timestamp=datetime.now()
            )
        
        return None
    
    def _check_anomalous_location(self, event: LoginEvent) -> Optional[SecurityAlert]:
        """
        规则2: 异地登录检测
        
        逻辑：
        - 对比登录地点与用户常用地点
        - 如果地点不在常用列表中，检查时间窗口
        - 如果短时间内从两地登录，产生告警
        """
        user_profile = self.user_profiles.get(event.username)
        
        if not user_profile:
            # 新用户，记录但不告警（需要学习期）
            return None
        
        # 检查是否在常用地点
        if event.geolocation in user_profile.usual_locations:
            return None
        
        # 地点异常，进一步判断
        alerts = []
        
        # 检查时间合理性（飞行时间）
        if user_profile.last_login_time:
            time_diff = (event.timestamp - user_profile.last_login_time).total_seconds() / 3600
            
            # 如果上次登录在2小时前，但地点跨洲了，这不合理
            if time_diff < 2 and event.geolocation not in user_profile.usual_locations:
                return SecurityAlert(
                    alert_id=self._gen_alert_id("IMPOSSIBLE_TRAVEL"),
                    rule_name="不可能的旅行-异地快速登录",
                    risk_level=RiskLevel.CRITICAL,
                    event=event,
                    description=(
                        f"账号 {event.username} 在 {time_diff:.1f} 小时内从两个不同地点登录，"
                        f"上次: {user_profile.last_login_time} ({event.geolocation})，"
                        f"本次: {event.geolocation}，物理上不可能"
                    ),
                    suggested_action="立即锁定账号，联系用户确认，检查是否存在会话劫持",
                    timestamp=event.timestamp
                )
        
        # 一般异地告警
        return SecurityAlert(
            alert_id=self._gen_alert_id("UNUSUAL_LOCATION"),
            rule_name="异地登录检测",
            risk_level=RiskLevel.MEDIUM,
            event=event,
            description=(
                f"账号 {event.username} 从非常用地点 {event.geolocation} 登录，"
                f"常用地点: {user_profile.usual_locations}"
            ),
            suggested_action="发送二次验证请求，记录设备指纹，通知用户确认",
            timestamp=event.timestamp
        )
    
    def _check_unusual_time(self, event: LoginEvent) -> Optional[SecurityAlert]:
        """
        规则3: 非工作时间登录检测
        
        逻辑：
        - 检查登录时间是否在工作时间范围内
        - 管理员账号在非工作时间登录触发更高级别告警
        """
        user_profile = self.user_profiles.get(event.username)
        
        if not user_profile:
            return None
        
        current_hour = event.timestamp.hour
        
        start_hour, end_hour = user_profile.usual_time_range
        
        # 在工作时间内，不告警
        if start_hour <= current_hour <= end_hour:
            return None
        
        # 非工作时间，但属于管理员角色，高危
        if user_profile.role == "admin":
            return SecurityAlert(
                alert_id=self._gen_alert_id("ADMIN_OFF_HOURS"),
                rule_name="管理员非工作时间登录",
                risk_level=RiskLevel.HIGH,
                event=event,
                description=(
                    f"管理员账号 {event.username} 在非工作时间 "
                    f"({current_hour}点) 登录，正常时间: {start_hour}-{end_hour}点"
                ),
                suggested_action="确认是否为计划内维护操作，要求管理员提供工单号，否则锁定账号",
                timestamp=event.timestamp
            )
        
        # 普通用户，中危
        return SecurityAlert(
            alert_id=self._gen_alert_id("OFF_HOURS_LOGIN"),
            rule_name="非工作时间登录",
            risk_level=RiskLevel.MEDIUM,
            event=event,
            description=(
                f"账号 {event.username} 在 {current_hour}点 登录，"
                f"正常工作时间: {start_hour}-{end_hour}点"
            ),
            suggested_action="记录日志，如需紧急访问要求主管审批",
            timestamp=event.timestamp
        )
    
    def _check_proxy_abuse(self, event: LoginEvent) -> Optional[SecurityAlert]:
        """
        规则4: 代理/VPN滥用检测
        
        逻辑：
        - 检查IP是否属于已知代理/VPN服务商
        - 检查IP是否被多个账号共享使用
        """
        if not event.is_proxy and not event.is_vpn:
            return None
        
        # 检查是否被多账号共享
        shared_accounts = [
            e.username for e in self.event_window.get(event.username, [])
            if e.source_ip == event.source_ip and e.username != event.username
        ]
        
        # 同时检查全局共享情况
        global_shared = False
        for username, events in self.event_window.items():
            if username == event.username:
                continue
            if any(e.source_ip == event.source_ip for e in events):
                global_shared = True
                break
        
        if event.is_proxy and global_shared:
            return SecurityAlert(
                alert_id=self._gen_alert_id("SHARED_PROXY"),
                rule_name="共享代理IP登录",
                risk_level=RiskLevel.HIGH,
                event=event,
                description=(
                    f"账号 {event.username} 通过代理IP {event.source_ip} 登录，"
                    f"该IP同时被其他账号使用，疑似共享代理/跳板机"
                ),
                suggested_action="要求使用企业VPN，检查代理IP归属，考虑封禁该IP段",
                timestamp=event.timestamp
            )
        
        if event.is_vpn:
            return SecurityAlert(
                alert_id=self._gen_alert_id("VPN_LOGIN"),
                rule_name="VPN登录记录",
                risk_level=RiskLevel.LOW,
                event=event,
                description=f"账号 {event.username} 通过VPN登录，IP: {event.source_ip}",
                suggested_action="记录日志备查",
                timestamp=event.timestamp
            )
        
        return None
    
    def _check_account_activity(self, event: LoginEvent) -> Optional[SecurityAlert]:
        """
        规则5: 账号异常活跃检测
        
        逻辑：
        - 检测登录成功后立即的高频操作
        - 批量下载、权限变更等敏感操作
        """
        # 这里简化处理，实际可以结合操作日志
        # 示例：登录成功后5分钟内发起超过20个API请求
        recent_events = [
            e for e in self.event_window.get(event.username, [])
            if e.timestamp > event.timestamp - timedelta(minutes=5)
        ]
        
        if len(recent_events) > 20 and event.login_result:
            return SecurityAlert(
                alert_id=self._gen_alert_id("HIGH_FREQUENCY"),
                rule_name="账号异常活跃",
                risk_level=RiskLevel.HIGH,
                event=event,
                description=(
                    f"账号 {event.username} 登录成功后5分钟内产生 {len(recent_events)} 个事件，"
                    f"疑似自动化脚本或数据爬取"
                ),
                suggested_action="暂停账号活动，人工复核操作内容，检查是否有数据泄露",
                timestamp=event.timestamp
            )
        
        return None
    
    def _update_user_profile(self, event: LoginEvent):
        """更新用户行为档案"""
        profile = self.user_profiles.get(event.username)
        if not profile:
            # 新用户，创建基础档案
            self.user_profiles[event.username] = UserProfile(
                username=event.username,
                usual_locations=[event.geolocation] if event.geolocation else [],
                usual_ips=[event.source_ip],
                usual_time_range=(9, 18),
                usual_devices=[event.device_fingerprint] if event.device_fingerprint else [],
                role="user"
            )
            return
        
        # 增量更新常用地点（滑动窗口）
        if event.geolocation and event.geolocation not in profile.usual_locations:
            profile.usual_locations.append(event.geolocation)
            if len(profile.usual_locations) > 5:
                profile.usual_locations = profile.usual_locations[-5:]
        
        if event.source_ip not in profile.usual_ips:
            profile.usual_ips.append(event.source_ip)
            if len(profile.usual_ips) > 10:
                profile.usual_ips = profile.usual_ips[-10:]
        
        profile.last_login_time = event.timestamp
    
    def _gen_alert_id(self, prefix: str) -> str:
        """生成唯一告警ID"""
        unique = hashlib.md5(
            f"{prefix}_{datetime.now().isoformat()}_{id(self)}".encode()
        ).hexdigest()[:8]
        return f"ALERT-{prefix}-{unique.upper()}"
    
    def get_alert_summary(self) -> dict:
        """获取告警统计摘要"""
        summary = {
            "total_alerts": len(self.alerts),
            "by_risk_level": defaultdict(int),
            "by_rule": defaultdict(int),
            "recent_events": len(sum(self.event_window.values(), []))
        }
        
        for alert in self.alerts:
            summary["by_risk_level"][alert.risk_level.value] += 1
            summary["by_rule"][alert.rule_name] += 1
        
        return dict(summary)

---

5.3 规则配置示例（JSON格式）

除了代码实现，规则引擎通常也支持配置化规则：

{
  "rule_engine_version": "2.1",
  "rules": [
    {
      "id": "RULE-001",
      "name": "暴力破解检测",
      "enabled": true,
      "conditions": {
        "operator": "AND",
        "checks": [
          {
            "field": "login_result",
            "operator": "==",
            "value": false
          },
          {
            "field": "failure_count",
            "operator": ">=",
            "value": 5,
            "time_window": "10m"
          }
        ]
      },
      "actions": [
        {
          "type": "alert",
          "level": "high",
          "notify": ["security-team@company.com"]
        },
        {
          "type": "lock_account",
          "duration": "30m"
        }
      ]
    },
    {
      "id": "RULE-002",
      "name": "异地登录检测",
      "enabled": true,
      "conditions": {
        "operator": "AND",
        "checks": [
          {
            "field": "geolocation",
            "operator": "not_in",
            "value": "{{user.usual_locations}}"
          },
          {
            "field": "time_since_last_login",
            "operator": "<",
            "value": "2h"
          }
        ]
      },
      "actions": [
        {
          "type": "alert",
          "level": "critical",
          "notify": ["security-team@company.com", "incident-response@company.com"]
        },
        {
          "type": "force_mfa",
          "channel": "sms"
        }
      ]
    },
    {
      "id": "RULE-003",
      "name": "管理员非工作时间登录",
      "enabled": true,
      "conditions": {
        "operator": "AND",
        "checks": [
          {
            "field": "user_role",
            "operator": "==",
            "value": "admin"
          },
          {
            "field": "hour",
            "operator": "not_in",
            "value": [9, 10, 11, 12, 13, 14, 15, 16, 17, 18]
          }
        ]
      },
      "actions": [
        {
          "type": "alert",
          "level": "high",
          "notify": ["it-manager@company.com"]
        },
        {
          "type": "require_approval",
          "approvers": ["cto@company.com"]
        }
      ]
    }
  ]
}

---

六、完整实战：从部署到响应

6.1 部署架构

┌─────────────────────────────────────────────────────────┐
│                    企业网络架构                          │
├─────────────┬─────────────┬─────────────┬───────────────┤
│  Web服务器  │  API网关    │  数据库     │   堡垒机      │
│  (认证日志) │  (访问日志) │  (操作日志) │  (运维日志)   │
└──────┬──────┴──────┬──────┴──────┬──────┴───────┬───────┘
       │             │             │              │
       └─────────────┴──────┬──────┴──────────────┘
                            │
                   ┌────────▼────────┐
                   │   日志收集器     │
                   │  (Fluentd/Logstash)│
                   └────────┬────────┘
                            │
                   ┌────────▼────────┐
                   │   规则引擎       │
                   │  (实时规则评估)  │
                   └────────┬────────┘
                            │
              ┌─────────────┼─────────────┐
              │             │             │
      ┌───────▼──────┐ ┌───▼────┐ ┌─────▼──────┐
      │  告警通知     │ │ 自动处置│ │  日志归档   │
      │ (邮件/钉钉)  │ │(封禁/锁)│ │ (ELK/SIEM) │
      └──────────────┘ └────────┘ └────────────┘

---

6.2 实战演示：一个完整的检测流程

让我们用刚才的规则引擎，模拟一次真实的攻击场景：

def simulate_attack_scenario():
    """模拟一次完整的攻击检测场景"""
    
    engine = LoginSecurityRuleEngine()
    
    # 注册用户基线
    admin_profile = UserProfile(
        username="zhang_admin",
        usual_locations=["北京-朝阳区"],
        usual_ips=["10.0.1.0/24", "192.168.1.0/24"],
        usual_time_range=(9, 18),
        usual_devices=["device_mac_book_pro"],
        role="admin"
    )
    engine.register_user_profile(admin_profile)
    
    employee_profile = UserProfile(
        username="li_employee",
        usual_locations=["北京-海淀区"],
        usual_ips=["10.0.2.0/24"],
        usual_time_range=(9, 18),
        usual_devices=["device_dell_laptop"],
        role="user"
    )
    engine.register_user_profile(employee_profile)
    
    print("=" * 60)
    print("场景1: 正常登录")
    print("=" * 60)
    
    normal_login = LoginEvent(
        event_id="EVT-001",
        timestamp=datetime.now(),
        username="li_employee",
        source_ip="10.0.2.15",
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
        login_result=True,
        auth_method="password",
        geolocation="北京-海淀区",
        device_fingerprint="device_dell_laptop",
        is_vpn=False,
        is_proxy=False
    )
    
    alerts = engine.evaluate(normal_login)
    print(f"告警数量: {len(alerts)}")
    if alerts:
        for a in alerts:
            print(f"  - [{a.risk_level.value}] {a.rule_name}: {a.description}")
    else:
        print("  无告警，登录正常 ✓")
    
    print("\n" + "=" * 60)
    print("场景2: 暴力破解攻击")
    print("=" * 60)
    
    # 模拟5次失败登录
    for i in range(5):
        fail_event = LoginEvent(
            event_id=f"EVT-BF-{i}",
            timestamp=datetime.now() - timedelta(minutes=5-i),
            username="zhang_admin",
            source_ip="203.0.113.45",  # 境外IP
            user_agent="python-requests/2.28",
            login_result=False,
            auth_method="password",
            geolocation="俄罗斯-莫斯科",
            device_fingerprint=None,
            is_vpn=False,
            is_proxy=False
        )
        alerts = engine.evaluate(fail_event)
        if alerts:
            for a in alerts:
                print(f"  - [{a.risk_level.value}] {a.rule_name}")
                print(f"    描述: {a.description}")
                print(f"    建议: {a.suggested_action}")
    
    print("\n" + "=" * 60)
    print("场景3: 异地登录（不可能的旅行）")
    print("=" * 60)
    
    # 先在北京登录
    beijing_login = LoginEvent(
        event_id="EVT-002",
        timestamp=datetime.now() - timedelta(hours=1),
        username="zhang_admin",
        source_ip="10.0.1.100",
        user_agent="Mozilla/5.0",
        login_result=True,
        auth_method="password",
        geolocation="北京-朝阳区",
        device_fingerprint="device_mac_book_pro",
        is_vpn=False,
        is_proxy=False
    )
    engine.evaluate(beijing_login)
    
    # 1小时后从境外登录
    foreign_login = LoginEvent(
        event_id="EVT-003",
        timestamp=datetime.now(),
        username="zhang_admin",
        source_ip="198.51.100.23",
        user_agent="Mozilla/5.0 (Windows NT 10.0)",
        login_result=True,
        auth_method="password",
        geolocation="美国-洛杉矶",
        device_fingerprint="device_windows_pc",
        is_vpn=False,
        is_proxy=False
    )
    
    alerts = engine.evaluate(foreign_login)
    for a in alerts:
        print(f"  - [{a.risk_level.value}] {a.rule_name}")
        print(f"    描述: {a.description}")
        print(f"    建议: {a.suggested_action}")
    
    print("\n" + "=" * 60)
    print("场景4: 管理员非工作时间登录")
    print("=" * 60)
    
    night_login = LoginEvent(
        event_id="EVT-004",
        timestamp=datetime.now().replace(hour=3, minute=15),
        username="zhang_admin",
        source_ip="10.0.1.100",
        user_agent="Mozilla/5.0",
        login_result=True,
        auth_method="password",
        geolocation="北京-朝阳区",
        device_fingerprint="device_mac_book_pro",
        is_vpn=False,
        is_proxy=False
    )
    
    alerts = engine.evaluate(night_login)
    for a in alerts:
        print(f"  - [{a.risk_level.value}] {a.rule_name}")
        print(f"    描述: {a.description}")
        print(f"    建议: {a.suggested_action}")
    
    print("\n" + "=" * 60)
    print("告警统计")
    print("=" * 60)
    
    summary = engine.get_alert_summary()
    print(f"总告警数: {summary['total_alerts']}")
    print(f"按风险等级:")
    for level, count in summary['by_risk_level'].items():
        print(f"  {level}: {count} 条")
    print(f"按规则:")
    for rule, count in summary['by_rule'].items():
        print(f"  {rule}: {count} 条")


if __name__ == "__main__":
    simulate_attack_scenario()

运行结果示例

============================================================
场景1: 正常登录
============================================================
告警数量: 0
  无告警，登录正常 ✓

============================================================
场景2: 暴力破解攻击
============================================================
  - [高] 暴力破解检测-多次失败
    描述: 账号 zhang_admin 在10分钟内失败登录 5 次，疑似暴力破解
    建议: 锁定账号并通知管理员，检查该IP是否在其他系统有异常行为

============================================================
场景3: 异地登录（不可能的旅行）
============================================================
  - [危急] 不可能的旅行-异地快速登录
    描述: 账号 zhang_admin 在 1.0 小时内从两个不同地点登录，上次: 北京-朝阳区，本次: 美国-洛杉矶，物理上不可能
    建议: 立即锁定账号，联系用户确认，检查是否存在会话劫持

============================================================
场景4: 管理员非工作时间登录
============================================================
  - [高] 管理员非工作时间登录
    描述: 管理员账号 zhang_admin 在非工作时间 (3点) 登录，正常时间: 9-18点
    建议: 确认是否为计划内维护操作，要求管理员提供工单号，否则锁定账号

============================================================
告警统计
============================================================
总告警数: 4
按风险等级:
  高: 2 条
  危急: 1 条
  中: 1 条
按规则:
  暴力破解检测-多次失败: 1 条
  不可能的旅行-异地快速登录: 1 条
  管理员非工作时间登录: 1 条
  非工作时间登录: 1 条

---

七、规则引擎的进阶玩法

7.1 机器学习 + 规则引擎：最佳组合

纯规则引擎有个问题：规则是死的，攻击是活的。

所以业界最佳实践是：规则引擎负责确定性检测，机器学习负责异常发现。

┌─────────────────────────────────────────┐
│           安全监控决策链                  │
├─────────────────────────────────────────┤
│                                         │
│  登录事件 ──► 规则引擎（快速匹配）       │
│               │                          │
│               ├─ 命中规则 ──► 立即响应   │
│               │                          │
│               └─ 未命中 ──► 机器学习模型 │
│                              │           │
│                        正常 ──► 放行     │
│                        异常 ──► 人工审核 │
│                                         │
└─────────────────────────────────────────┘

机器学习模型可以做的事：

1. UEBA（用户实体行为分析）：学习每个用户的正常行为模式，发现偏离
2. 集群检测：发现异常的IP聚集行为（僵尸网络特征）
3. 序列建模：识别攻击链（侦察→暴力破解→权限提升→数据窃取）

---

7.2 自动化响应（SOAR）

规则引擎检测到异常后，可以自动触发响应：

class AutomatedResponse:
    """自动化响应执行器"""
    
    def __init__(self, engine: LoginSecurityRuleEngine):
        self.engine = engine
    
    def execute_response(self, alert: SecurityAlert):
        """根据告警类型执行响应"""
        
        response_map = {
            "BRUTE_FORCE": self._handle_brute_force,
            "IMPOSSIBLE_TRAVEL": self._handle_impossible_travel,
            "ADMIN_OFF_HOURS": self._handle_admin_off_hours,
            "UNUSUAL_LOCATION": self._handle_unusual_location,
        }
        
        handler = response_map.get(alert.rule_name.split("-")[0])
        if handler:
            handler(alert)
    
    def _handle_brute_force(self, alert: SecurityAlert):
        """暴力破解响应"""
        print(f"[自动响应] 执行暴力破解处置:")
        print(f"  1. 锁定账号: {alert.event.username}")
        print(f"  2. 封禁IP: {alert.event.source_ip}")
        print(f"  3. 通知安全团队")
        print(f"  4. 保留证据日志")
    
    def _handle_impossible_travel(self, alert: SecurityAlert):
        """不可能旅行响应"""
        print(f"[自动响应] 执行不可能旅行处置:")
        print(f"  1. 立即终止所有活跃会话")
        print(f"  2. 强制重置密码")
        print(f"  3. 启用紧急MFA验证")
        print(f"  4. 上报安全事件")
        print(f"  5. 启动 Incident Response 流程")
    
    def _handle_admin_off_hours(self, alert: SecurityAlert):
        """管理员非工作时间响应"""
        print(f"[自动响应] 执行管理员非工作时间响应:")
        print(f"  1. 发送审批请求给主管")
        print(f"  2. 记录完整操作日志")
        print(f"  3. 限制敏感操作权限")

---

八、中小企业如何落地：分步指南

第一步：盘点资产（1天）

先搞清楚你家有什么：

• 多少台服务器
• 哪些系统有登录功能
• 日志存在哪里
• 有多少管理员账号

不要跳这一步，很多公司跳过了盘点，直接买产品，结果发现日志收集都搞不定。

---

第二步：定义基线（3-5天）

为每个用户/角色建立行为基线：

• 正常登录时间
• 常用IP段
• 常用地点
• 常用设备

可以用一周的观察数据来建立基线。

---

第三步：配置规则（2-3天）

从最核心的规则开始：

优先级	规则	说明
P0	暴力破解检测	最直接的威胁，优先级最高
P0	账号锁定策略	防止密码被爆破后持续访问
P1	异地登录检测	检测账号被盗
P1	管理员特殊监控	管理员权限大，需要更严的规则
P2	非工作时间登录	减少误报，先做监控
P2	代理IP检测	识别跳板机

---

第四步：灰度上线（1周）

不要直接开自动响应！

第一阶段：只告警，不处置。观察一周，看看规则有没有误报。

如果误报太多，调整阈值：

• 失败次数从5次调到10次
• 时间窗口从10分钟调到30分钟
• 异地判断加个”常用地点”白名单

---

第五步：全量上线（持续优化）

确认规则有效后，开启自动响应。同时建立定期review机制：

• 每周查看误报/漏报
• 每月调整规则阈值
• 每季度更新攻击模式库

---

九、常见坑，踩一个就亏一万

坑一：规则太严，业务跑不动

有个客户把”异地登录”规则设成”任何非本地IP都告警”，结果销售天天出差，告警每天几十条，最后运维直接关了规则。

建议：区分”告警”和”阻断”，先观察再处置。

---

坑二：只看登录，不看操作

登录正常不代表操作正常。很多内鬼是正常登录，然后批量导出数据。

建议：规则引擎要覆盖整个操作链，不只是登录。

---

坑三：忽视日志质量

规则引擎再好，日志采集不到也是白搭。

检查清单：

• [ ] 登录成功/失败都有日志
• [ ] 日志包含时间、IP、账号、操作
• [ ] 日志时间同步（NTP）
• [ ] 日志保留时长≥180天

---

坑四：规则不更新，变成摆设

攻击手法在变，规则也要变。去年有效的规则，今年可能就不行了。

建议：建立规则review机制，每月至少检查一次规则效果。

---

十、一个真实案例：某电商公司如何防住0.5元损失

有个做跨境电商的朋友，他们的规则引擎配置了以下规则：

1. 异地登录+管理员账号 = 立即锁定
2. 同一IP短时间多账号登录 = 封禁IP
3. 非工作时间导出数据 = 需要二次审批

某天凌晨2点，规则引擎检测到：

• IP 45.33.xx.xx（境外）
• 10分钟内尝试登录15个不同账号
• 其中一个账号是财务主管

系统自动：

1. 封禁了该IP
2. 锁定了被尝试的账号
3. 通知了安全团队

后来查证，这是一个僵尸网络在扫账号。因为规则生效快，攻击者没有成功登录任何账号。

如果没有这个规则引擎，后果可能是：攻击者用撞库来的密码登录财务账号，修改收款信息，导致货款流入骗子账户。

---

十一、规则引擎 vs 其他安全方案，怎么选？

方案	优点	缺点	适用场景
规则引擎	响应快、可解释、配置灵活	需要手动维护规则	已知威胁、中小企业
机器学习	发现未知威胁、自适应	误报率高、需大量数据	大型企业、有安全团队
云端WAF	开箱即用、免维护	依赖供应商、规则不透明	Web应用防护
SIEM	日志集中、分析能力强	昂贵、配置复杂	中大型企业

我的建议：

中小企业先上规则引擎，成本最低、见效最快。等积累了足够的日志数据和攻击案例，再考虑引入机器学习。

---

十二、动手试试：你的第一套规则

如果你决定开始搭建，可以从这三条规则起步：

# 极简版规则定义
RULES = [
    {
        "name": "密码错误超过5次锁定",
        "condition": lambda event: event.login_result == False 
                                   and event.failed_count >= 5,
        "action": "lock_account"
    },
    {
        "name": "境外IP登录告警",
        "condition": lambda event: event.is_foreign_ip,
        "action": "alert"
    },
    {
        "name": "管理员非工作时间登录",
        "condition": lambda event: event.role == "admin" 
                                   and event.hour not in range(9, 19),
        "action": "alert_and_notify"
    }
]

先让这三条跑起来，观察一周，再逐步增加。

---

十三、总结：规则引擎不是银弹，但是个好起点

回到开头那个问题——为什么中小企业总被按在地上摩擦？

因为很多公司把安全当成买产品，而不是建能力。

规则引擎的核心价值，是把安全逻辑显式化、可维护、可迭代。它让你知道：

• 什么行为被监控
• 什么规则触发了告警
• 为什么触发
• 怎么响应

这些能力，比任何昂贵的设备都重要。

---

最后送一句话：安全不是一次性项目，是个持续迭代的过程。规则引擎是你最好的工具之一，因为它让你能快速试错、快速调整、快速验证。

你的企业，可以从今天开始，写第一条规则。