4：L的强化学习安全决策：蓝队的智能响应系统

import numpy as np
import gym
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env

# 定义安全决策环境
class SecurityEnv(gym.Env):
    def __init__(self):
        super(SecurityEnv, self).__init__()
        # 定义状态空间（网络流量特征、系统状态等）
        self.observation_space = gym.spaces.Box(low=0, high=1, shape=(10,), dtype=np.float32)
        # 定义动作空间（防御措施）
        self.action_space = gym.spaces.Discrete(5)  # 5种可能的防御动作
        # 初始化环境状态
        self.state = np.random.random(10)
        self.episode_length = 0
        self.max_episode_length = 100
    
    def reset(self):
        self.state = np.random.random(10)
        self.episode_length = 0
        return self.state
    
    def step(self, action):
        # 模拟动作执行
        # 这里简化处理，实际中需要根据动作类型执行相应的防御措施
        
        # 计算奖励
        # 假设动作0是最佳防御措施，其他动作效果依次递减
        if action == 0:
            reward = 1.0
        elif action == 1:
            reward = 0.7
        elif action == 2:
            reward = 0.5
        elif action == 3:
            reward = 0.3
        else:
            reward = 0.1
        
        # 模拟状态变化
        self.state = np.random.random(10)
        
        # 检查是否结束
        self.episode_length += 1
        done = self.episode_length >= self.max_episode_length
        
        return self.state, reward, done, {}

# 创建环境
env = SecurityEnv()

# 训练PPO模型
model = PPO('MlpPolicy', env, verbose=1)
model.learn(total_timesteps=10000)

# 测试模型
obs = env.reset()
done = False
total_reward = 0

while not done:
    action, _states = model.predict(obs, deterministic=True)
    obs, reward, done, info = env.step(action)
    total_reward += reward
    print(f"动作: {action}, 奖励: {reward}")

print(f"总奖励: {total_reward}")

def calculate_reward(state, action, next_state, attack_detected):
    """计算奖励函数"""
    # 基础奖励
    base_reward = 0.0
    
    # 安全效果奖励
    if attack_detected:
        if action == 0:  # 最佳防御动作
            security_reward = 1.0
        elif action == 1:
            security_reward = 0.7
        elif action == 2:
            security_reward = 0.5
        else:
            security_reward = 0.2
    else:
        # 没有攻击时，避免误报
        if action == 4:  # 无操作
            security_reward = 0.5
        else:
            security_reward = -0.1
    
    # 响应时间奖励（假设动作0响应时间最短）
    response_time_penalty = action * 0.05
    
    # 资源消耗奖励（假设动作0资源消耗最少）
    resource_penalty = action * 0.03
    
    # 综合奖励
    total_reward = base_reward + security_reward - response_time_penalty - resource_penalty
    
    return total_reward

# 测试奖励函数
state = np.random.random(10)
action = 0
next_state = np.random.random(10)
attack_detected = True

reward = calculate_reward(state, action, next_state, attack_detected)
print(f"奖励: {reward}")

import networkx as nx

class NetworkEnvironment:
    def __init__(self):
        # 构建网络拓扑
        self.network = nx.Graph()
        # 添加节点（网络设备）
        self.network.add_nodes_from(["firewall", "router", "server1", "server2", "client1", "client2"])
        # 添加边（网络连接）
        self.network.add_edges_from([("firewall", "router"), ("router", "server1"), 
                                    ("router", "server2"), ("router", "client1"), ("router", "client2")])
        
        # 初始化节点状态
        self.node_states = {
            "firewall": {"status": "normal", "load": 0.3},
            "router": {"status": "normal", "load": 0.4},
            "server1": {"status": "normal", "load": 0.6},
            "server2": {"status": "normal", "load": 0.5},
            "client1": {"status": "normal", "load": 0.2},
            "client2": {"status": "normal", "load": 0.2}
        }
    
    def simulate_attack(self, attack_type, target):
        """模拟攻击"""
        if attack_type == "DDoS":
            # 模拟DDoS攻击，增加目标节点的负载
            self.node_states[target]["load"] += 0.5
            if self.node_states[target]["load"] > 1.0:
                self.node_states[target]["status"] = "compromised"
        elif attack_type == "malware":
            # 模拟恶意软件攻击，直接将目标节点标记为被攻击
            self.node_states[target]["status"] = "compromised"
    
    def get_state(self):
        """获取环境状态"""
        # 将网络状态转换为向量
        state = []
        for node in ["firewall", "router", "server1", "server2", "client1", "client2"]:
            state.append(1.0 if self.node_states[node]["status"] == "compromised" else 0.0)
            state.append(self.node_states[node]["load"])
        return np.array(state)

# 测试环境模型
env = NetworkEnvironment()
print("初始状态:", env.get_state())

# 模拟攻击
env.simulate_attack("DDoS", "server1")
print("攻击后状态:", env.get_state())