# Detecting Anomalous Authentication Patterns

> 使用用户和实体行为分析（UEBA）分析、统计基线和机器学习模型检测异常认证模式，识别不可能旅行、撞库攻击、暴力破解、密码喷洒和账户被攻陷行为，覆盖所有认证日志来源。 适用于认证异常检测、登录行为分析、UEBA 实施或可疑登录调查等相关请求。

- Skill: `killvxk/detecting-anomalous-authentication-patterns` (Agent Skill, multi-file: 4 files)
- Install (CLI): `npx skillmds@latest add killvxk/detecting-anomalous-authentication-patterns`
- Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/detecting-anomalous-authentication-patterns/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- License: Apache-2.0
- Author: killvxk (https://skillmd.com/u/killvxk)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/killvxk/detecting-anomalous-authentication-patterns

---


# 检测异常认证模式

## 适用场景

- 安全运营需要通过认证日志分析识别被攻陷的账户
- 实施不可能旅行检测（Impossible Travel Detection），标记地理上不一致的登录
- 实时检测暴力破解（Brute Force）、密码喷洒（Password Spraying）和撞库（Credential Stuffing）攻击
- 为用户建立行为基线，识别表明账户被攻陷的偏差
- 将认证异常与威胁情报相关联，用于横向移动检测
- 调查来自 SIEM 或 IdP 的可疑登录活动告警

**不适用于**针对单次失败登录的静态规则告警；异常检测需要跨时间和实体维度的统计基线，以减少误报。

## 前置条件

- 认证日志来源（Azure AD/Entra ID 登录日志、Okta 系统日志、Active Directory 事件日志 4624/4625/4648/4768/4771）
- SIEM 平台（Splunk、Microsoft Sentinel、Elastic SIEM），至少包含 90 天基线数据
- 用于基于位置的异常检测的 GeoIP 数据库（MaxMind GeoLite2 或 IP2Location）
- Python 3.9+（需安装 pandas、scikit-learn、scipy）用于自定义分析
- 用户身份上下文（部门、角色、典型工作时间、位置）

## 工作流程

### 步骤 1：收集和规范化认证日志

从所有身份来源聚合认证事件：

```python
import pandas as pd
import json
from datetime import datetime, timedelta
from collections import defaultdict

# 从多个来源解析认证日志
def normalize_auth_logs(log_source, raw_logs):
    """将认证事件规范化为统一格式。"""
    normalized = []

    for event in raw_logs:
        if log_source == "azure_ad":
            normalized.append({
                "timestamp": event["createdDateTime"],
                "user": event["userPrincipalName"],
                "source_ip": event["ipAddress"],
                "location": {
                    "city": event.get("location", {}).get("city"),
                    "state": event.get("location", {}).get("state"),
                    "country": event.get("location", {}).get("countryOrRegion"),
                    "lat": event.get("location", {}).get("geoCoordinates", {}).get("latitude"),
                    "lon": event.get("location", {}).get("geoCoordinates", {}).get("longitude")
                },
                "result": "success" if event["status"]["errorCode"] == 0 else "failure",
                "failure_reason": event["status"].get("failureReason", ""),
                "app": event.get("appDisplayName", "Unknown"),
                "device": event.get("deviceDetail", {}).get("operatingSystem", "Unknown"),
                "browser": event.get("deviceDetail", {}).get("browser", "Unknown"),
                "mfa_result": event.get("authenticationDetails", [{}])[0].get("succeeded", None),
                "risk_level": event.get("riskLevelDuringSignIn", "none"),
                "client_app": event.get("clientAppUsed", "Unknown"),
                "source": "azure_ad"
            })
        elif log_source == "okta":
            normalized.append({
                "timestamp": event["published"],
                "user": event["actor"]["alternateId"],
                "source_ip": event["client"]["ipAddress"],
                "location": {
                    "city": event["client"].get("geographicalContext", {}).get("city"),
                    "state": event["client"].get("geographicalContext", {}).get("state"),
                    "country": event["client"].get("geographicalContext", {}).get("country"),
                    "lat": event["client"].get("geographicalContext", {}).get("geolocation", {}).get("lat"),
                    "lon": event["client"].get("geographicalContext", {}).get("geolocation", {}).get("lon")
                },
                "result": "success" if event["outcome"]["result"] == "SUCCESS" else "failure",
                "failure_reason": event["outcome"].get("reason", ""),
                "app": event.get("target", [{}])[0].get("displayName", "Unknown"),
                "device": event["client"].get("device", "Unknown"),
                "browser": event["client"].get("userAgent", {}).get("browser", "Unknown"),
                "source": "okta"
            })
        elif log_source == "windows_ad":
            normalized.append({
                "timestamp": event["TimeCreated"],
                "user": event["TargetUserName"],
                "source_ip": event.get("IpAddress", ""),
                "location": None,  # 需要 GeoIP 增强
                "result": "success" if event["EventId"] in [4624, 4648] else "failure",
                "failure_reason": event.get("FailureReason", ""),
                "logon_type": event.get("LogonType", ""),
                "source": "windows_ad"
            })

    return pd.DataFrame(normalized)

# 为缺少位置信息的 Windows AD 日志添加 GeoIP 数据
import geoip2.database

def enrich_geoip(df, geoip_db_path="/opt/geoip/GeoLite2-City.mmdb"):
    """为缺少位置信息的事件添加地理位置数据。"""
    reader = geoip2.database.Reader(geoip_db_path)

    for idx, row in df.iterrows():
        if row["location"] is None and row["source_ip"]:
            try:
                response = reader.city(row["source_ip"])
                df.at[idx, "location"] = {
                    "city": response.city.name,
                    "country": response.country.iso_code,
                    "lat": response.location.latitude,
                    "lon": response.location.longitude
                }
            except Exception:
                pass

    reader.close()
    return df
```

### 步骤 2：检测不可能旅行异常

识别地理位置上不可能的登录：

```python
from math import radians, sin, cos, sqrt, atan2
from datetime import datetime

def haversine_distance(lat1, lon1, lat2, lon2):
    """计算两点之间的大圆距离（单位：千米）。"""
    R = 6371  # 地球半径（千米）

    lat1, lon1, lat2, lon2 = map(radians, [lat1, lon1, lat2, lon2])
    dlat = lat2 - lat1
    dlon = lon2 - lon1

    a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlon/2)**2
    c = 2 * atan2(sqrt(a), sqrt(1-a))

    return R * c

def detect_impossible_travel(df, max_speed_kmh=900):
    """
    检测不可能旅行事件——用户在物理上不可能的时间内
    从两个不同地点进行认证。

    max_speed_kmh：最大现实旅行速度（900 千米/小时 ≈ 商业航班）
    """
    alerts = []

    # 按用户和时间戳排序
    df_sorted = df.sort_values(["user", "timestamp"])

    for user, user_events in df_sorted.groupby("user"):
        successful_events = user_events[user_events["result"] == "success"]

        for i in range(1, len(successful_events)):
            prev = successful_events.iloc[i-1]
            curr = successful_events.iloc[i]

            # 如果缺少位置数据则跳过
            if not prev.get("location") or not curr.get("location"):
                continue
            if not prev["location"].get("lat") or not curr["location"].get("lat"):
                continue

            # 计算距离和时间差
            distance_km = haversine_distance(
                prev["location"]["lat"], prev["location"]["lon"],
                curr["location"]["lat"], curr["location"]["lon"]
            )

            time_diff = (pd.Timestamp(curr["timestamp"]) -
                        pd.Timestamp(prev["timestamp"])).total_seconds() / 3600

            if time_diff <= 0:
                continue

            required_speed = distance_km / time_diff

            # 如果所需速度超过最大现实旅行速度则标记
            if required_speed > max_speed_kmh and distance_km > 100:
                alerts.append({
                    "alert_type": "IMPOSSIBLE_TRAVEL",
                    "severity": "HIGH",
                    "user": user,
                    "timestamp": curr["timestamp"],
                    "details": {
                        "location_1": f"{prev['location']['city']}, {prev['location']['country']}",
                        "location_2": f"{curr['location']['city']}, {curr['location']['country']}",
                        "time_1": prev["timestamp"],
                        "time_2": curr["timestamp"],
                        "distance_km": round(distance_km, 1),
                        "time_hours": round(time_diff, 2),
                        "required_speed_kmh": round(required_speed, 1),
                        "source_ip_1": prev["source_ip"],
                        "source_ip_2": curr["source_ip"]
                    }
                })

    return alerts

# 运行不可能旅行检测
travel_alerts = detect_impossible_travel(auth_df)
print(f"不可能旅行告警：{len(travel_alerts)}")
for alert in travel_alerts:
    print(f"  [{alert['severity']}] {alert['user']}: "
          f"{alert['details']['location_1']} -> {alert['details']['location_2']} "
          f"（{alert['details']['distance_km']} 千米，{alert['details']['time_hours']} 小时）")
```

### 步骤 3：检测暴力破解和密码喷洒

识别认证日志中的凭据攻击模式：

```python
from collections import Counter

def detect_brute_force(df, threshold_failures=10, window_minutes=10):
    """
    检测暴力破解攻击：在短时间窗口内对
    单个账户的多次失败尝试。
    """
    alerts = []
    failed = df[df["result"] == "failure"].copy()
    failed["timestamp"] = pd.to_datetime(failed["timestamp"])

    for user, user_fails in failed.groupby("user"):
        user_fails_sorted = user_fails.sort_values("timestamp")

        # 滑动窗口分析
        for i, row in user_fails_sorted.iterrows():
            window_start = row["timestamp"]
            window_end = window_start + timedelta(minutes=window_minutes)

            window_events = user_fails_sorted[
                (user_fails_sorted["timestamp"] >= window_start) &
                (user_fails_sorted["timestamp"] <= window_end)
            ]

            if len(window_events) >= threshold_failures:
                source_ips = window_events["source_ip"].unique()
                alerts.append({
                    "alert_type": "BRUTE_FORCE",
                    "severity": "HIGH",
                    "user": user,
                    "timestamp": str(window_start),
                    "details": {
                        "failed_attempts": len(window_events),
                        "window_minutes": window_minutes,
                        "source_ips": list(source_ips),
                        "distributed": len(source_ips) > 1,
                        "failure_reasons": dict(Counter(window_events["failure_reason"]))
                    }
                })
                break  # 每次检测每个用户只生成一个告警

    return alerts

def detect_password_spray(df, threshold_users=10, window_minutes=30):
    """
    检测密码喷洒：在短时间窗口内从同一来源
    对许多不同账户的失败登录（每个用户 1-2 次尝试）。
    """
    alerts = []
    failed = df[df["result"] == "failure"].copy()
    failed["timestamp"] = pd.to_datetime(failed["timestamp"])

    for source_ip, ip_events in failed.groupby("source_ip"):
        ip_events_sorted = ip_events.sort_values("timestamp")

        for i, row in ip_events_sorted.iterrows():
            window_start = row["timestamp"]
            window_end = window_start + timedelta(minutes=window_minutes)

            window_events = ip_events_sorted[
                (ip_events_sorted["timestamp"] >= window_start) &
                (ip_events_sorted["timestamp"] <= window_end)
            ]

            unique_users = window_events["user"].nunique()
            attempts_per_user = len(window_events) / unique_users if unique_users > 0 else 0

            # 密码喷洒：针对大量用户，每个用户尝试次数少
            if unique_users >= threshold_users and attempts_per_user <= 3:
                # 检查是否有成功登录（账户被攻陷）
                success_after = df[
                    (df["source_ip"] == source_ip) &
                    (df["result"] == "success") &
                    (pd.to_datetime(df["timestamp"]) > window_start) &
                    (pd.to_datetime(df["timestamp"]) < window_end + timedelta(hours=1))
                ]

                alerts.append({
                    "alert_type": "PASSWORD_SPRAY",
                    "severity": "CRITICAL" if len(success_after) > 0 else "HIGH",
                    "timestamp": str(window_start),
                    "details": {
                        "source_ip": source_ip,
                        "targeted_users": unique_users,
                        "total_attempts": len(window_events),
                        "avg_attempts_per_user": round(attempts_per_user, 1),
                        "window_minutes": window_minutes,
                        "successful_logins_after": len(success_after),
                        "compromised_accounts": list(success_after["user"].unique()) if len(success_after) > 0 else []
                    }
                })
                break

    return alerts

# 运行检测
brute_force_alerts = detect_brute_force(auth_df)
spray_alerts = detect_password_spray(auth_df)
print(f"暴力破解告警：{len(brute_force_alerts)}")
print(f"密码喷洒告警：{len(spray_alerts)}")
```

### 步骤 4：建立行为基线并检测偏差

创建用户行为档案并标记统计异常：

```python
import numpy as np
from scipy import stats
from sklearn.ensemble import IsolationForest

def build_user_baseline(df, user, lookback_days=90):
    """为特定用户建立行为基线。"""
    user_events = df[df["user"] == user].copy()
    user_events["timestamp"] = pd.to_datetime(user_events["timestamp"])
    user_events["hour"] = user_events["timestamp"].dt.hour
    user_events["day_of_week"] = user_events["timestamp"].dt.dayofweek

    baseline = {
        "user": user,
        "typical_hours": {
            "start": int(user_events["hour"].quantile(0.05)),
            "end": int(user_events["hour"].quantile(0.95)),
            "mean": float(user_events["hour"].mean()),
            "std": float(user_events["hour"].std())
        },
        "typical_days": list(user_events["day_of_week"].mode().values),
        "typical_ips": list(user_events["source_ip"].value_counts().head(10).index),
        "typical_locations": list(
            user_events["location"].apply(
                lambda x: x.get("country") if isinstance(x, dict) else None
            ).dropna().value_counts().head(5).index
        ),
        "typical_apps": list(user_events["app"].value_counts().head(10).index),
        "typical_devices": list(user_events["device"].value_counts().head(5).index),
        "avg_daily_logins": float(
            user_events.groupby(user_events["timestamp"].dt.date).size().mean()
        ),
        "std_daily_logins": float(
            user_events.groupby(user_events["timestamp"].dt.date).size().std()
        ),
        "failure_rate": float(
            (user_events["result"] == "failure").mean()
        )
    }

    return baseline

def detect_behavioral_anomalies(event, baseline):
    """将新认证事件与用户基线进行比较。"""
    anomalies = []
    event_time = pd.Timestamp(event["timestamp"])

    # 非工作时间登录检测
    hour = event_time.hour
    if baseline["typical_hours"]["std"] > 0:
        z_score = abs(hour - baseline["typical_hours"]["mean"]) / baseline["typical_hours"]["std"]
        if z_score > 2.5:
            anomalies.append({
                "type": "OFF_HOURS_LOGIN",
                "severity": "MEDIUM",
                "detail": f"在 {hour}:00 登录（基线：{baseline['typical_hours']['start']}:00-{baseline['typical_hours']['end']}:00）",
                "z_score": round(z_score, 2)
            })

    # 新来源 IP
    if event["source_ip"] not in baseline["typical_ips"]:
        anomalies.append({
            "type": "NEW_SOURCE_IP",
            "severity": "MEDIUM",
            "detail": f"从未知 IP 登录：{event['source_ip']}"
        })

    # 新国家/地区
    if event.get("location") and isinstance(event["location"], dict):
        country = event["location"].get("country")
        if country and country not in baseline["typical_locations"]:
            anomalies.append({
                "type": "NEW_COUNTRY",
                "severity": "HIGH",
                "detail": f"从新国家/地区登录：{country}"
            })

    # 新应用程序
    if event.get("app") and event["app"] not in baseline["typical_apps"]:
        anomalies.append({
            "type": "NEW_APPLICATION",
            "severity": "LOW",
            "detail": f"访问新应用程序：{event['app']}"
        })

    # 新设备
    if event.get("device") and event["device"] not in baseline["typical_devices"]:
        anomalies.append({
            "type": "NEW_DEVICE",
            "severity": "MEDIUM",
            "detail": f"从新设备登录：{event['device']}"
        })

    # 仅工作日用户的周末登录
    if event_time.dayofweek >= 5 and 5 not in baseline["typical_days"] and 6 not in baseline["typical_days"]:
        anomalies.append({
            "type": "WEEKEND_LOGIN",
            "severity": "LOW",
            "detail": f"检测到周末登录（典型工作日：{baseline['typical_days']}）"
        })

    return anomalies

def isolation_forest_anomaly_detection(df):
    """使用孤立森林（Isolation Forest）进行多变量异常检测。"""
    # 特征工程
    features_df = df.copy()
    features_df["timestamp"] = pd.to_datetime(features_df["timestamp"])
    features_df["hour"] = features_df["timestamp"].dt.hour
    features_df["day_of_week"] = features_df["timestamp"].dt.dayofweek
    features_df["is_failure"] = (features_df["result"] == "failure").astype(int)

    # 编码分类特征
    features_df["ip_frequency"] = features_df.groupby("source_ip")["source_ip"].transform("count")
    features_df["user_frequency"] = features_df.groupby("user")["user"].transform("count")

    feature_columns = ["hour", "day_of_week", "is_failure", "ip_frequency", "user_frequency"]
    X = features_df[feature_columns].fillna(0)

    # 训练孤立森林
    model = IsolationForest(
        n_estimators=200,
        contamination=0.01,  # 预期 1% 的异常率
        random_state=42,
        n_jobs=-1
    )
    features_df["anomaly_score"] = model.fit_predict(X)
    features_df["anomaly_probability"] = model.score_samples(X)

    # 提取异常（标记为 -1）
    anomalies = features_df[features_df["anomaly_score"] == -1]

    return anomalies.sort_values("anomaly_probability")
```

### 步骤 5：实施 SIEM 检测规则

为常见认证攻击模式部署检测规则：

```yaml
# 用于认证异常检测的 Splunk SPL 查询

# 1. 暴力破解检测
# 名称：认证暴力破解 - 多次失败登录
# 严重级别：高
brute_force_spl: |
  index=auth sourcetype IN ("azure:aad:signin", "okta:im:log", "WinEventLog:Security")
  (result="failure" OR EventCode=4625)
  | bin _time span=10m
  | stats count as failed_attempts dc(src_ip) as unique_ips
    values(src_ip) as source_ips
    latest(_time) as last_attempt
    by user _time
  | where failed_attempts >= 10
  | eval alert_type=if(unique_ips > 3, "分布式暴力破解", "标准暴力破解")

# 2. 密码喷洒检测
# 名称：密码喷洒攻击 - 相同来源针对多个用户
# 严重级别：严重
password_spray_spl: |
  index=auth sourcetype IN ("azure:aad:signin", "okta:im:log")
  result="failure"
  | bin _time span=30m
  | stats dc(user) as targeted_users count as total_attempts
    values(user) as users_targeted
    by src_ip _time
  | where targeted_users >= 10
  | eval attempts_per_user = round(total_attempts / targeted_users, 1)
  | where attempts_per_user <= 3
  | eval severity=if(targeted_users > 50, "CRITICAL", "HIGH")

# 3. 不可能旅行检测
# 名称：不可能旅行 - 地理上不一致的登录
# 严重级别：高
impossible_travel_spl: |
  index=auth result="success"
  | iplocation src_ip
  | sort user _time
  | streamstats current=f last(lat) as prev_lat last(lon) as prev_lon
    last(_time) as prev_time last(City) as prev_city last(Country) as prev_country
    by user
  | where isnotnull(prev_lat) AND isnotnull(lat)
  | eval distance_km = 6371 * 2 * asin(sqrt(
      pow(sin((lat - prev_lat) * pi() / 360), 2) +
      cos(prev_lat * pi() / 180) * cos(lat * pi() / 180) *
      pow(sin((lon - prev_lon) * pi() / 360), 2)))
  | eval time_hours = (_time - prev_time) / 3600
  | eval required_speed = distance_km / time_hours
  | where required_speed > 900 AND distance_km > 100

# 4. 撞库检测
# 名称：撞库攻击 - 大量失败登录后部分成功
# 严重级别：严重
credential_stuffing_spl: |
  index=auth
  | bin _time span=1h
  | stats count(eval(result="failure")) as failures
    count(eval(result="success")) as successes
    dc(user) as unique_users
    dc(src_ip) as unique_ips
    by src_ip _time
  | where failures > 100 AND successes > 0 AND unique_users > 20
  | eval success_rate = round(successes / (failures + successes) * 100, 2)
  | where success_rate < 5
```

### 步骤 6：关联和评分认证异常

将多个检测信号合并为风险评分：

```python
def calculate_auth_risk_score(user, alerts, baseline):
    """
    计算认证事件的综合风险评分。
    将多个异常信号与加权评分相结合。
    """
    score = 0
    risk_factors = []

    weights = {
        "IMPOSSIBLE_TRAVEL": 40,
        "PASSWORD_SPRAY": 35,
        "BRUTE_FORCE": 30,
        "CREDENTIAL_STUFFING": 35,
        "NEW_COUNTRY": 25,
        "OFF_HOURS_LOGIN": 15,
        "NEW_SOURCE_IP": 10,
        "NEW_DEVICE": 10,
        "NEW_APPLICATION": 5,
        "WEEKEND_LOGIN": 5,
        "MFA_BYPASS": 45,
        "LEGACY_PROTOCOL": 20
    }

    for alert in alerts:
        alert_type = alert.get("type") or alert.get("alert_type")
        weight = weights.get(alert_type, 10)

        # 根据严重级别调整权重
        severity_multiplier = {
            "CRITICAL": 2.0,
            "HIGH": 1.5,
            "MEDIUM": 1.0,
            "LOW": 0.5
        }
        severity = alert.get("severity", "MEDIUM")
        adjusted_weight = weight * severity_multiplier.get(severity, 1.0)

        score += adjusted_weight
        risk_factors.append({
            "factor": alert_type,
            "weight": adjusted_weight,
            "detail": alert.get("detail", alert.get("details", ""))
        })

    # 将评分规范化到 0-100
    normalized_score = min(100, score)

    # 确定风险级别
    if normalized_score >= 80:
        risk_level = "CRITICAL"
        recommended_action = "立即暂停账户并展开调查"
    elif normalized_score >= 60:
        risk_level = "HIGH"
        recommended_action = "强制 MFA 重新注册并通知 SOC"
    elif normalized_score >= 40:
        risk_level = "MEDIUM"
        recommended_action = "要求升级认证"
    elif normalized_score >= 20:
        risk_level = "LOW"
        recommended_action = "监控并记录用于趋势分析"
    else:
        risk_level = "INFORMATIONAL"
        recommended_action = "无需采取行动"

    return {
        "user": user,
        "risk_score": normalized_score,
        "risk_level": risk_level,
        "recommended_action": recommended_action,
        "risk_factors": sorted(risk_factors, key=lambda x: x["weight"], reverse=True),
        "timestamp": datetime.utcnow().isoformat()
    }
```

## 核心概念

| 术语 | 定义 |
|------|------------|
| **不可能旅行（Impossible Travel）** | 认证异常，用户在物理上不可能旅行的时间内从两个地理位置相距甚远的地方登录 |
| **密码喷洒（Password Spraying）** | 凭据攻击，对大量账户尝试少量常用密码，以规避账户锁定阈值 |
| **撞库（Credential Stuffing）** | 自动化攻击，使用从数据泄露中获取的用户名/密码组合，尝试非法访问账户 |
| **UEBA（用户和实体行为分析）** | 使用机器学习和统计分析建立行为基线并检测偏差的技术 |
| **行为基线（Behavioral Baseline）** | 用户正常认证模式的统计档案，包括典型时间、位置、设备和应用程序 |
| **孤立森林（Isolation Forest）** | 无监督机器学习算法，通过孤立与大多数数据点不同的观测值来检测异常 |
| **风险评分（Risk Score）** | 聚合多个异常信号并进行加权评分的综合数值，用于优先处理认证威胁 |

## 工具与系统

- **Microsoft Sentinel UEBA**：云原生 SIEM，内置实体行为分析，用于 Azure AD 和多云认证异常检测
- **Exabeam Advanced Analytics**：UEBA 平台，使用机器学习进行用户会话分析和自动化威胁时间线构建
- **Splunk UBA**：Splunk 的行为分析附加组件，提供预构建的认证异常模型和风险评分
- **Elastic SIEM ML Jobs**：Elastic Stack 中用于认证日志分析的机器学习异常检测作业

## 常见场景

### 场景：密码喷洒后检测被攻陷的高管账户

**场景背景**：SOC 发现来自云 VPS IP 地址的失败认证尝试激增，针对 200 多个账户。两小时后，一名高管账户从同一 IP 范围显示成功认证，随后创建了邮箱规则并发生数据外泄。

**方法**：
1. 在该时间段内运行密码喷洒检测，识别所有被针对的账户
2. 将被针对的账户与来自相关 IP 范围的后续成功登录进行交叉参照
3. 为高管账户建立行为基线，标记所有偏差
4. 检查高管最后一次合法登录与攻击者会话之间的不可能旅行
5. 识别攻陷后活动：邮箱规则、文件下载、委托访问更改
6. 计算综合风险评分，结合密码喷洒、新 IP、非工作时间登录和新设备信号
7. 触发自动响应：强制终止会话、禁用账户、通知经理

**常见陷阱**：
- 依赖单一信号检测（仅看失败登录）会遗漏成功的喷洒结果
- 当用户在多个 IdP 中有账户时，未跨身份提供商进行关联
- 未考虑合法 VPN IP 变化或出差的静态阈值
- 忽视喷洒窗口关闭后的成功认证（攻击者可能在使用凭据前等待）

## 输出格式

```
认证异常检测报告
=========================================
分析期间：     2026-02-01 至 2026-02-24
总认证事件数：  2,847,392
监控用户数：    3,847
告警来源：      Azure AD、Okta、Windows AD

威胁检测概要
密码喷洒攻击：  3 次
暴力破解攻击：  12 次
不可能旅行：    8 次
撞库攻击：      1 次
行为异常：      47 次

高风险账户
[CRITICAL] j.smith@corp.com     评分：92
  - 不可能旅行：芝加哥 -> 莫斯科（7,876 千米，0.5 小时）
  - 密码喷洒目标，随后成功登录
  - 新设备和浏览器指纹
  - 非工作时间访问 SharePoint 和电子邮件
  行动：账户已暂停，SOC 已启动调查

[HIGH] m.johnson@corp.com       评分：67
  - 从新国家/地区登录（巴西）
  - 新来源 IP 不匹配 VPN 范围
  - 访问 HR 应用程序超出正常模式
  行动：要求 MFA 重新注册，已通知经理

[MEDIUM] a.williams@corp.com    评分：38
  - 周末 UTC 03:00 登录
  - 新设备（Linux，通常使用 Windows）
  行动：已应用升级认证

攻击活动详情
密码喷洒活动 #1：
  来源：         185.220.101.x/24（Tor 出口节点）
  目标用户：     247
  成功率：       0.8%（2 个账户被攻陷）
  被攻陷账户：   j.smith@corp.com、r.davis@corp.com
  持续时间：     45 分钟
  模式：         每个用户 2 次尝试，3 秒间隔
```