2026-10-23 22:43:19 网络安全文章来源：ZONE.CI 全球网 0 阅读模式

cover_image

Ai养蛊:让钓鱼邮件和反钓鱼邮件系统打一架

秋名山上的小柠
秋名山上的小柠

蚁景网络安全

2026年6月30日 17:40
湖南

在小说阅读器读本章

去阅读

mab

多臂老虎机，又称为mab

同一个环境，动作，状态下有可能返回1，有可能返回0

也就是说环境反馈它不是一个固定的值

可以假设为有五个函数，也就是相当于五种反馈，第一个函数返回1的概率是20％，返回0的概率是80％

代码实现：

ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineimport numpy as npimport pandas as pd
class MultiArmedBandit:    def __init__(self, n_arms, true_rewards):        self.n_arms = n_arms        self.true_rewards = true_rewards        self.estimates = np.zeros(n_arms)  # 每个臂的奖励估计        self.action_counts = np.zeros(n_arms)  # 每个臂被选择的次数
    def select_arm(self, epsilon):        if np.random.rand() < epsilon:            return np.random.randint(self.n_arms)  # 探索        else:            return np.argmax(self.estimates)  # 开发
    def update_estimates(self, chosen_arm, reward):        self.action_counts[chosen_arm] += 1        # 更新奖励估计        self.estimates[chosen_arm] += (reward - self.estimates[chosen_arm]) / self.action_counts[chosen_arm]
def simulate_bandit(n_arms, true_rewards, n_rounds, epsilon):    bandit = MultiArmedBandit(n_arms, true_rewards)    rewards = np.zeros(n_rounds)    cumulative_rewards = np.zeros(n_rounds)
    for round in range(n_rounds):        chosen_arm = bandit.select_arm(epsilon)        reward = np.random.normal(true_rewards[chosen_arm], 1)  # 奖励是正态分布        bandit.update_estimates(chosen_arm, reward)        rewards[round] = reward        cumulative_rewards[round] = np.sum(rewards)
    return cumulative_rewards
# 参数设置n_arms = 5true_rewards = [1.0, 1.5, 2.0, 0.5, 1.2]  # 每个臂的真实奖励均值n_rounds = 1000epsilon = 0.1
cumulative_rewards = simulate_bandit(n_arms, true_rewards, n_rounds, epsilon)results_df = pd.DataFrame({    'Round': np.arange(1, n_rounds + 1),    'Cumulative Rewards': cumulative_rewards})
results_df

类定义：MultiArmedBandit

n\_arms 老虎机的数量

true\_rewards 每个臂的真实平均奖励

estimates 目前认为每个臂的平均回报是多少，初始全为0

action\_counts 记录每个臂被拉了多少次，用于更新均值

选择臂：select\_arm(self, epsilon)

然后定义一个随机数

以概率 ε 进行探索，也就是随机选一个臂，以概率 1 - ε 进行开发（选当前估计奖励最高的臂）。

比如说当 epsilon = 0.1：

10% 概率随机探索；
90% 概率选估计最好的那一个

更新估计值：update\_estimates()

R 是这次的实际奖励；N 是该臂被选过的次数；Q 是对该臂期望奖励的估计

模拟函数：simulate\_bandit()

初始化一个 MultiArmedBandit 实例；
进行多轮（n_rounds）实验；
每一轮：

用 select_arm() 决定拉哪一台机器；
根据真实均值 true_rewards[chosen_arm] 生成一个服从正态分布的奖励；
用 update_estimates() 更新估计；
记录当前的奖励和累计奖励

效果如图所示：

ucb

UCB算法是一种用于解决探索与利用问题的策略选择方法，广泛应用于多臂老虎机问题

其核心思想是通过估计每个选项的潜在收益来平衡探索新选项和利用已知最佳选项之间的权衡

基本原理 1. 探索与利用：探索：尝试新的选项以获取更多的信息，利用：选择当前已知的最佳选项以最大化收益

UCB值计算：对于每个选项，UCB算法计算一个上置信界值也就是UCB值，该值结合了成功率和探索因子

计算公式：

X\_i 是选项 i 的成功率,即平均收益; n 是当前总的尝试次数; n\_i 是选项 i 的尝试次数

第一项是指当前已知的平均成功率；第二项是指置信区间，也就是越没试过的策略，这项越大；比如说你去饭堂吃饭，吃过 10 次的店你知道它一般，但没吃过的店你可能会想试一试，这就是 UCB 的探索机制

应用场景 UCB算法广泛应用于在线广告推荐、A/B测试、动态定价、机器学习模型选择等领域，尤其是在需要实时决策和反馈的环境中

ucb的通俗解释：一个左撇子，用手拿东西的时候，用右手的概率是20% ，用左手的概率是80%由于第一次选择的时候左右都会选，但是概率不同，选择不同手的频率就会影响两边ubc（可以理解为Q表）的值那么我们就可以根据两边受频率影响的值动态调整我们是否选择高的那边的概率

防火墙策略

假设有五个防火墙策略，并且拦截攻击的成功率都不一致

但是在实际项目中，不用都写出成功率出来，毕竟只要知道哪个防火墙拦截的成功率高，那肯定优先选择那个防火墙

现在是不知道概率多少

ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineimport numpy as npimport pandas as pd
def check1(payload):    return np.random.rand() < 0.5  # 50%成功率
def check2(payload):    return np.random.rand() < 0.7  # 70%成功率
def check3(payload):    return np.random.rand() < 0.4  # 40%成功率
def check4(payload):    return np.random.rand() < 0.3  # 30%成功率
def check5(payload):    return np.random.rand() < 0.6  # 60%成功率
# 将所有检查函数放入列表中check_functions = [check1, check2, check3, check4, check5]
# 定义防火墙策略选择器类class FirewallPolicySelector:    def __init__(self, n_policies):        self.n_policies = n_policies        self.successes = np.zeros(n_policies)        self.attempts = np.zeros(n_policies)
    def select_policy(self):        total_attempts = np.sum(self.attempts)        if total_attempts == 0:            return np.random.randint(self.n_policies)  # 如果没有尝试过，随机选择        ucb_values = self.successes / (self.attempts + 1e-5) + np.sqrt(2 * np.log(total_attempts) / (self.attempts + 1e-5))        return np.argmax(ucb_values)  # 选择UCB值最高的策略
    def update(self, chosen_policy, success):        self.attempts[chosen_policy] += 1        self.successes[chosen_policy] += success
# 模拟防火墙策略优化过程def simulate_firewall(n_policies, n_rounds):    policy_selector = FirewallPolicySelector(n_policies)    results = []
    for round in range(n_rounds):        chosen_policy = policy_selector.select_policy()        payload = np.random.randint(0, 100)  # 生成随机攻击样本        success = check_functions[chosen_policy](payload)  # 使用选定的check函数        policy_selector.update(chosen_policy, success)        results.append((round + 1, chosen_policy, success))
    results_df = pd.DataFrame(results, columns=['轮次', '选择的策略', '成功拦截'])    return results_df
# 参数设置n_policies = len(check_functions)  # 策略数量n_rounds = 1000
# 运行模拟results_df = simulate_firewall(n_policies, n_rounds)
# 筛选出成功拦截的部分successful_results = results_df[results_df['成功拦截'] == 1]
# 输出每个策略的成功率print(\"\
每个策略的成功率：\")print(results_df.groupby('选择的策略')['成功拦截'].mean())
# 显示成功拦截的结果print(\"\
成功拦截的结果：\")print(successful_results)# 统计每个策略的选择次数policy_counts = results_df['选择的策略'].value_counts()
# 创建 DataFrame 显示所有策略及其选择次数result_df = pd.DataFrame({    '选择次数': policy_counts}).reset_index()
# 重命名列result_df.columns = ['选择的策略', '选择次数']
# 设置行标题result_df.index = [f'策略 {i+1}' for i in range(len(result_df))]result_df

防火墙策略选择器类 FirewallPolicySelector

n_policies: 策略数量;successes[i]: 第 i 个策略成功的次数;attempts[i]: 第 i 个策略被尝试的次数

策略选择核心 select\_policy()

这里用的ucb计算公式，在上述已贴出

模拟防火墙运行：simulate\_firewall()

循环共执行 n_rounds，比如 1000 轮：

选择一个策略，然后模拟生成攻击，接着判断是否成功拦截，最后更新策略统计

简单来说，这份代码就是模拟了一个基于UCB算法的自适应防火墙策略选择系统，它通过统计每个检测策略的历史成功率和尝试次数，自动在多轮攻击中选择最有效的策略，在“探索新方法”和“利用已知最优”之间取得平衡，最终趋向于选择拦截率最高的策略

效果如图：

其实还有其他场景也适合，比如说什么恶意代码识别，邮箱识别，毕竟是策略选择

邮件攻防

假设现在有个角色A 通过mba模型实现强化学习下的优化钓鱼邮件内容

还有一个角色B 通过Q-learning的方式实现强化学习下的钓鱼邮件内容识别

当然也可以换成一边是恶意软件，一边杀毒软件，做一个养蛊哈哈

整个流程就是攻击方不断发送不同类型的钓鱼邮件，防御方在识别的过程中逐渐学习，而攻击方也会记录哪些内容更容易成功，从而倾向选择这些高成功率内容

ounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineounter(lineimport numpy as npimport pandas as pd
class PhishingContentOptimizer:    def __init__(self, contents, phishing_probabilities, epsilon=0.1):        self.contents = contents  # 钓鱼邮件内容列表        self.phishing_probabilities = phishing_probabilities  # 各内容被识别为钓鱼邮件的概率        self.epsilon = epsilon  # 探索率        self.success_counts = np.zeros(len(contents))  # 各内容成功次数        self.total_counts = np.zeros(len(contents))  # 各内容尝试次数
    def select_content(self):        if np.random.rand() < self.epsilon:            return np.random.choice(self.contents)  # 随机选择        else:            success_rates = self.success_counts / (self.total_counts + 1e-5)  # 避免除零            return self.contents[np.argmax(success_rates)]  # 选择成功率最高的内容
    def update(self, chosen_content, success):        index = self.contents.index(chosen_content)        self.total_counts[index] += 1        if success:            self.success_counts[index] += 1
class QLearningPhishingDetector:    def __init__(self, actions, learning_rate=0.1, discount_factor=0.9, exploration_rate=1.0):        self.q_table = {}  # Q值表        self.actions = actions  # 可采取的动作        self.learning_rate = learning_rate  # 学习率        self.discount_factor = discount_factor  # 折扣因子        self.exploration_rate = exploration_rate  # 探索率        self.exploration_decay = 0.99  # 探索率衰减
    def get_action(self, state):        if state not in self.q_table:            self.q_table[state] = [0] * len(self.actions)        if np.random.rand() < self.exploration_rate:            return np.random.choice(self.actions)  # 探索        else:            return self.actions[np.argmax(self.q_table[state])]  # 利用
    def update_q_value(self, state, action, reward, next_state):        current_q = self.q_table[state]        max_future_q = max(self.q_table.get(next_state, [0] * len(self.actions)))        current_q[action] += self.learning_rate * (reward + self.discount_factor * max_future_q - current_q[action])  # 更新Q值
    def decay_exploration(self):        self.exploration_rate *= self.exploration_decay
# 示例钓鱼邮件内容及其被识别为钓鱼邮件的概率contents = [    \"您的账户存在异常，请立即验证。\",    \"恭喜您获得奖品，请点击链接领取。\",    \"重要通知：请更新您的账户信息。\",    \"您有新的消息，请查看。\",    \"系统升级，请确认您的信息。\",]
# 各内容被识别为钓鱼邮件的概率phishing_probabilities = {    contents[0]: 0.1,    contents[1]: 0.3,    contents[2]: 0.6,    contents[3]: 0.5,    contents[4]: 0.4,}
# 初始化角色A（内容优化器）optimizer = PhishingContentOptimizer(contents, phishing_probabilities)
# 初始化角色B（钓鱼邮件识别器）actions = [0, 1]  # 0: 正常邮件, 1: 钓鱼邮件detector = QLearningPhishingDetector(actions)
# 预训练阶段pretrain_steps = 50  # 预训练步骤数for _ in range(pretrain_steps):    chosen_content = np.random.choice(contents)  # 随机选择内容    action = detector.get_action(chosen_content)  # 识别邮件    # 根据内容的钓鱼概率判断    success = np.random.rand() < phishing_probabilities[chosen_content] if action == 1 else False    reward = 1 if action == 1 and success else -1  # 奖励机制    detector.update_q_value(chosen_content, action, reward, chosen_content)  # 更新Q值    detector.decay_exploration()  # 衰减探索率
# 模拟钓鱼攻击过程results = []
for _ in range(100):  # 模拟100次钓鱼攻击    chosen_content = optimizer.select_content()
    # 角色B识别邮件    action = detector.get_action(chosen_content)  # 识别邮件    results.append({        '选择的内容': chosen_content,        '识别结果': '钓鱼邮件' if action == 1 else '正常邮件'    })
# 统计识别结果的成功率for result in results:    if result['识别结果'] == '钓鱼邮件':        # 根据内容的钓鱼概率判断        success = np.random.rand() < phishing_probabilities[result['选择的内容']]    else:        success = False  # 正常邮件识别为钓鱼邮件的成功率为0
    # 更新角色A的成功与否    optimizer.update(result['选择的内容'], success)
    # 更新角色B的Q值    reward = 1 if action == 1 and success else -1  # 奖励机制    detector.update_q_value(result['选择的内容'], action, reward, result['选择的内容'])  # 更新Q值    detector.decay_exploration()  # 衰减探索率
# 转换为DataFrameresults_df = pd.DataFrame(results)
# 输出结果print(results_df)
# 统计每个内容的使用频率content_counts = results_df['选择的内容'].value_counts()most_used_content = content_counts.idxmax()most_used_count = content_counts.max()
# 筛选出使用最多的内容的结果most_used_results = results_df[results_df['选择的内容'] == most_used_content]
# 输出使用最多的内容print(f\"\
使用最多的内容: {most_used_content}, 使用次数: {most_used_count}\")print(\"\
使用最多内容的结果：\")print(most_used_results)
# 统计识别结果为正常邮件的百分比normal_email_count = results_df[results_df['识别结果'] == '正常邮件'].shape[0]total_count = results_df.shape[0]normal_email_percentage = (normal_email_count / total_count) * 100
print(f\"\
识别结果为正常邮件的百分比: {normal_email_percentage:.2f}%\")

钓鱼内容优化器 PhishingContentOptimizer

contents: 所有钓鱼邮件的模板内容；

phishing_probabilities: 每种内容被识别为钓鱼的概率，也就是被识破的难度；

epsilon: ε-贪婪算法中的“探索率”，比如 0.1 意味着 10% 概率随机探索；

success_counts: 各邮件“成功骗过检测”的次数；

total_counts: 每个内容被使用的次数

选择内容 select\_content

每轮发送邮件前，优化器根据历史成功率决定发哪种内容：

90% 概率选择成功率最高的邮件；
10% 概率随机选一个探索新的可能

这样攻击方会逐渐聚焦在最有效的邮件内容上

钓鱼邮件检测器 QLearningPhishingDetector

属性	含义
`q_table`	存储每个状态（邮件内容）对应的动作（判定）价值
`actions`	可选动作：0=正常邮件，1=钓鱼邮件
`learning_rate`	学习率 α，控制更新步幅
`discount_factor`	折扣因子 γ，考虑未来回报
`exploration_rate`	当前探索概率（从1开始逐步下降）
`exploration_decay`	探索衰减速率（每次减少到99%）