| name | detecting-anomalous-authentication-patterns |
| description | 使用用户和实体行为分析（UEBA）分析、统计基线和机器学习模型检测异常认证模式，识别不可能旅行、撞库攻击、暴力破解、密码喷洒和账户被攻陷行为，覆盖所有认证日志来源。 适用于认证异常检测、登录行为分析、UEBA 实施或可疑登录调查等相关请求。
|
| domain | cybersecurity |
| subdomain | identity-access-management |
| tags | ["UEBA","authentication-anomaly","impossible-travel","brute-force","credential-stuffing","behavioral-analytics"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
检测异常认证模式
适用场景
- 安全运营需要通过认证日志分析识别被攻陷的账户
- 实施不可能旅行检测（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：收集和规范化认证日志
从所有身份来源聚合认证事件：
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 ,
: event[].get(, ),
: event.get(, ),
: event.get(, {}).get(, ),
: event.get(, {}).get(, ),
: event.get(, [{}])[].get(, ),
: event.get(, ),
: event.get(, ),
:
})
log_source == :
normalized.append({
: event[],
: event[][],
: event[][],
: {
: event[].get(, {}).get(),
: event[].get(, {}).get(),
: event[].get(, {}).get(),
: event[].get(, {}).get(, {}).get(),
: event[].get(, {}).get(, {}).get()
},
: event[][] == ,
: event[].get(, ),
: event.get(, [{}])[].get(, ),
: event[].get(, ),
: event[].get(, {}).get(, ),
:
})
log_source == :
normalized.append({
: event[],
: event[],
: event.get(, ),
: ,
: event[] [, ] ,
: event.get(, ),
: event.get(, ),
:
})
pd.DataFrame(normalized)
geoip2.database
():
reader = geoip2.database.Reader(geoip_db_path)
idx, row df.iterrows():
row[] row[]:
:
response = reader.city(row[])
df.at[idx, ] = {
: response.city.name,
: response.country.iso_code,
: response.location.latitude,
: response.location.longitude
}
Exception:
reader.close()
df
步骤 2：检测不可能旅行异常
识别地理位置上不可能的登录：
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"):
prev[].get() curr[].get():
distance_km = haversine_distance(
prev[][], prev[][],
curr[][], curr[][]
)
time_diff = (pd.Timestamp(curr[]) -
pd.Timestamp(prev[])).total_seconds() /
time_diff <= :
required_speed = distance_km / time_diff
required_speed > max_speed_kmh distance_km > :
alerts.append({
: ,
: ,
: user,
: curr[],
: {
: ,
: ,
: prev[],
: curr[],
: (distance_km, ),
: (time_diff, ),
: (required_speed, ),
: prev[],
: curr[]
}
})
alerts
travel_alerts = detect_impossible_travel(auth_df)
()
alert travel_alerts:
(
)
步骤 3：检测暴力破解和密码喷洒
识别认证日志中的凭据攻击模式：
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),
: (source_ips) > ,
: (Counter(window_events[]))
}
})
alerts
():
alerts = []
failed = df[df[] == ].copy()
failed[] = pd.to_datetime(failed[])
source_ip, ip_events failed.groupby():
ip_events_sorted = ip_events.sort_values()
i, row ip_events_sorted.iterrows():
window_start = row[]
window_end = window_start + timedelta(minutes=window_minutes)
window_events = ip_events_sorted[
(ip_events_sorted[] >= window_start) &
(ip_events_sorted[] <= window_end)
]
unique_users = window_events[].nunique()
attempts_per_user = (window_events) / unique_users unique_users >
unique_users >= threshold_users attempts_per_user <= :
success_after = df[
(df[] == source_ip) &
(df[] == ) &
(pd.to_datetime(df[]) > window_start) &
(pd.to_datetime(df[]) < window_end + timedelta(hours=))
]
alerts.append({
: ,
: (success_after) > ,
: (window_start),
: {
: source_ip,
: unique_users,
: (window_events),
: (attempts_per_user, ),
: window_minutes,
: (success_after),
: (success_after[].unique()) (success_after) > []
}
})
alerts
brute_force_alerts = detect_brute_force(auth_df)
spray_alerts = detect_password_spray(auth_df)
()
()
步骤 4：建立行为基线并检测偏差
创建用户行为档案并标记统计异常：
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") (x, )
).dropna().value_counts().head().index
),
: (user_events[].value_counts().head().index),
: (user_events[].value_counts().head().index),
: (
user_events.groupby(user_events[].dt.date).size().mean()
),
: (
user_events.groupby(user_events[].dt.date).size().std()
),
: (
(user_events[] == ).mean()
)
}
baseline
():
anomalies = []
event_time = pd.Timestamp(event[])
hour = event_time.hour
baseline[][] > :
z_score = (hour - baseline[][]) / baseline[][]
z_score > :
anomalies.append({
: ,
: ,
: ,
: (z_score, )
})
event[] baseline[]:
anomalies.append({
: ,
: ,
:
})
event.get() (event[], ):
country = event[].get()
country country baseline[]:
anomalies.append({
: ,
: ,
:
})
event.get() event[] baseline[]:
anomalies.append({
: ,
: ,
:
})
event.get() event[] baseline[]:
anomalies.append({
: ,
: ,
:
})
event_time.dayofweek >= baseline[] baseline[]:
anomalies.append({
: ,
: ,
:
})
anomalies
():
features_df = df.copy()
features_df[] = pd.to_datetime(features_df[])
features_df[] = features_df[].dt.hour
features_df[] = features_df[].dt.dayofweek
features_df[] = (features_df[] == ).astype()
features_df[] = features_df.groupby()[].transform()
features_df[] = features_df.groupby()[].transform()
feature_columns = [, , , , ]
X = features_df[feature_columns].fillna()
model = IsolationForest(
n_estimators=,
contamination=,
random_state=,
n_jobs=-
)
features_df[] = model.fit_predict(X)
features_df[] = model.score_samples(X)
anomalies = features_df[features_df[] == -]
anomalies.sort_values()
步骤 5：实施 SIEM 检测规则
为常见认证攻击模式部署检测规则：
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, "分布式暴力破解", "标准暴力破解")
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")
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
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：关联和评分认证异常
将多个检测信号合并为风险评分：
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({
: alert_type,
: adjusted_weight,
: alert.get(, alert.get(, ))
})
normalized_score = (, score)
normalized_score >= :
risk_level =
recommended_action =
normalized_score >= :
risk_level =
recommended_action =
normalized_score >= :
risk_level =
recommended_action =
normalized_score >= :
risk_level =
recommended_action =
:
risk_level =
recommended_action =
{
: user,
: normalized_score,
: risk_level,
: recommended_action,
: (risk_factors, key= x: x[], reverse=),
: 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 范围显示成功认证，随后创建了邮箱规则并发生数据外泄。
方法：
- 在该时间段内运行密码喷洒检测，识别所有被针对的账户
- 将被针对的账户与来自相关 IP 范围的后续成功登录进行交叉参照
- 为高管账户建立行为基线，标记所有偏差
- 检查高管最后一次合法登录与攻击者会话之间的不可能旅行
- 识别攻陷后活动：邮箱规则、文件下载、委托访问更改
- 计算综合风险评分，结合密码喷洒、新 IP、非工作时间登录和新设备信号
- 触发自动响应：强制终止会话、禁用账户、通知经理
常见陷阱：
- 依赖单一信号检测（仅看失败登录）会遗漏成功的喷洒结果
- 当用户在多个 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 秒间隔
