背景
Breach & Attack Simulation(BAS)作为一种前沿的安全技术,能够通过模拟真实攻击场景来度量企业的安全水位。可以验证现有安全感知和防护措施的有效性,从而确保企业安全能力的稳定性和持续提升。因此,BAS对于企业的整体安全策略具有非常重要的价值。
BAS评估用例需要投入资源开发,重复建设检测/防御策略+ BAS的评估用例对于企业成本过高。
然而,传统的BAS方案需要大量的测试用例来保证其对安全检测和防护效果的度量能力,这往往需要投入极大的开发成本。每一个测试用例的设计、实现和调试都需要耗费大量的时间和人力资源,使得BAS的实施变得复杂且成本高昂。通过Multi-Agent+LLM能力,我们可以尝试自动化的生成BAS用例。这将显著降低开发成本,提高测试用例的生成效率,从而较低成本的利用BAS来评估企业的安全防护能力。
具体实现
工作分解
LLM并不能胜任处理复杂的项目,所以我们可以将复杂的工作的分解,分别设置专业的Agent来处理细分的任务。将项目分解为单Agent+LLM能够胜任的任务需要遵循如下原则:
- 分解任务是定义结果,而不是拆分行为
- 互斥:不同的任务需要保持orthogonal
- 有明确约束
针对攻击模拟,对工作的任务完成分解后,会基于分解的任务对角色进行定义,同时明确其目标、约束和能力。最终角色列表如下:
Requirements Analyst
Goal: To analyze ATT&CK matrix Techniques, Tactics, and Procedure Examples, and define the necessary components for simulating attack scenarios.
Constraints: Ensure comprehensive understanding of the ATT&CK matrix and the specific APT behaviors. Communicate requirements clearly to the next roles.
Security Architect
Goal: To design the technical implementation of attack scenarios, specifying the required entities, environments, and dependencies.
Constraints: Use simple tools (e.g., nc, nmap) and common programming languages (e.g., Python) or open-source libraries (e.g., requests). Ensure the design is practical and efficient.
Product Architect
Goal: To design a concise, usable, complete software
Constraints: Ensure the architecture remains simple and utilize appropriate open-source libraries.
Project Manager
Goal: To develop a comprehensive task list based on the requirements and architecture, assigning roles and managing task dependencies and priorities effectively.
Constraints: Follow the requirements and architectural design closely. Ensure clear communication and precise task definitions.
SRE
Goal: To deploy and configure the necessary environments as specified by the Security Architect, ensuring operational readiness.
Constraints: Maintain security and stability of the deployed environments. Ensure configurations meet the specified requirements accurately.
Developer
Goal: To develop clean, maintainable, and efficient code for the attack scenarios based on the architecture and task assignments.
Constraints: Adhere to coding standards and best practices. Ensure the code is modular, readable, and aligns with the specified design.
Penetration Tester
Goal: To execute the simulated attack scenarios in the deployed environments, validating their success based on the defined verification methods.
Constraints: Follow the implementation plan provided by the Security Architect. Ensure thorough validation and accurate execution of the attack scenarios.
Tools实现
在拆解后的任务分配给Agent后,不同的Role需要相应的Tool来实现与环境的互动。如SRE的需要具备从Web搜索依赖工具的下载路径,和云平台交互创建测试环境,以及在实例上执行命令部署环境。
Tools借助Function Calling实现,在调用LLM时传递Function的描述,LLM根据Prompt会返回Action指定调用合适的函数并传递合适的参数,最后由Agent基于Action对函数发起调用完成交互。
SOP通信
Multi-Agent框架中Agent之间的交互以及Agent与LLM之间的交互通常通过自然语言传递。自然语言在传递信息和指令时容易产生歧义和丢失信息,而人类在企业管理和运营过程中已经创造一套比较好的最佳实践SOP。将SOP的思路应用在Agent之间以及Agent-LLM之间的通信有如下优势:
- 便于解析:自然语言难以解析,SOP化后形成固定格式,便于解析
- 减少错误:SOP具备格式和定义能够更精准的传递信息,避免出错;SOP显式的传递约束条件,能够更好的避免产生歧义
- 会话传递:自然语言容易丢失context,SOP能够显式的传递上下文
如下是通过Requirements Analyst通过SOP向Security Architect传递需求的案例。其中Context是Requirements Analyst基于ATT&CK矩阵的Procedure Example分析出的攻击目标和步骤,format example部分是希望LLM产出遵循的格式,nodes部分是对格式中字段的类型和内容的解释,constraint是对产出内容的约束条件,action是对LLM的需求。
## context
{"Original Requirements": {"Tactic": "Defense Evasion", "Technique": "Hidden Users", "Name": "Dragonfly", "Procedure Example": "Dragonfly has modified the Registry to hide created user accounts.(Citation: US-CERT TA18-074A)"}, "Roles": {"Attacker": "Modify the Registry to hide user accounts", "Victim": "Host with Registry modified by Attacker"}, "Steps In Operation": ["Attacker modifies the Registry on Victim\'s host to hide created user accounts"], "Verification Checklist": ["User accounts are not visible in the list of users on Victim\'s host", "Registry keys associated with hiding users are modified on Victim\'s host"]}
## format example
{
"environment":{
"Attacker A":{
"OS" : "ubuntu 22.04",
"Additional Requirements" : "Pre-installed Python environment"
},
"Victim A": {
"OS" : "ubuntu 22.04",
"Additional Requirements" : "Pre-installed Python environment"
},
"Victim B": {
"OS" : "ubuntu 22.04",
"Additional Requirements" : "Pre-installed Netcat"
},
},
"Configure": {
"Attacker A":{
"verification_file_path": {
"Value": "UnDetermined",
"Generation Logic": "Randomly generate paths based on Victim A's file system, determined after deployment",
"Description": "Files created by Attacker A"
},
"cs_listen_port": {
"Value": "18080",
"Generation Logic": "Randomly pick a value from the port range",
"Description": "listen to receive reverse shell"
},
"scanned_ip": {
"Value": "UnDetermined",
"Generation Logic": "Determined after deployment",
"Description": "Victim B's IP address"
}
},
"Victim A":{
"cs_listen_port": {
"Value": "18080",
"Generation Logic": "Randomly pick a value from the port range",
"Description": "listen to receive reverse shell"
},
"cs_ip": {
"Value": "UnDetermined",
"Generation Logic": "Determined after deployment",
"Description": "Attacker A's IP address"
}
},
"Victim B":{
"victim_listen_port": {
"Value": "8080",
"Generation Logic": "Randomly pick a value from the port range",
"Description": "listening for port scans by Victim A"
}
}
},
"resource development" :[
{
"Name" : "reverse_client",
"Product Goals": [
"perform reverse shel to remote address for security testing",
"perform port scanning. Server mode can listen to ports, accept reverse connections, and send commands for execution.",
"Port scanning using tcp connect"
],
"Requirement Pool": [
"The main code ...",
"Implement of reverse shell ...",
"Implement of port scan ..."
],
"Configure" :{
"connected IP" : "{cs_ip}",
"connected port" : "{cs_listen_port}"
},
"Anything UNCLEAR": ""
},
{
"Name" : "reverse_server",
"Product Goals": [
"accept reverse connections from victim",
"send commands for execution.",
"control victim to perform port scan"
],
"Requirement Pool": [
"The main code ...",
"Implement of reverse shell accept...",
"Implement of remote control ..."
],
"Configure" :{
"listen port" : "{cs_listen_port}",
"scanned ip" : "{scanned_ip}"
},
"Anything UNCLEAR": ""
}
],
"resource obtain" : {
"xxx", "download from github",
"Netcat", "Installation via package manager"
},
"Environmental deployment" : [
"deploy reverse_server on Attacker A",
"deploy reverse_client on Victim A",
"deploy Netcat on Victim B"
],
"Steps In Operation": [
"Attcker A use reverse_server to listen to port {cs_listen_port}"
"Victim A use reverse_client to establish a reverse shell connection to Attacker A ({cs_ip}, {cs_listen_port})",
"After Attacker A receives the reverse shell connection from Victim A, it uses reverse_server to create a file ({verification_file_path}) on Victim A to prove the attack was successful",
"Victim B use Netcat to listen to port {victim_listen_port}",
"Attacker A controls Victim A to perform port scan on Victim B ({scanned_ip}) via reverse_client"
],
"Verification Checklist":[
"{verification_file_path} exist on Victim A",
"detect port {victim_listen_port} open on Victim B ({scanned_ip})"
]
}
## nodes: "<node>: <type> # <instruction>"
-environment: typing.Dict[str, typing.Dict[str,str]] # Define the OS and additional environment requirements for each role, clearly specifying the pre-installed software needed.When choosing an environment, you need to consider an environment that is more suitable for creating on the cloud, more general, and more convenient for executing commands.
-configure: typing.Dict[str, typing.Dict[str, typing.Dict[str, str]]] # Detail the specific settings or configurations for each entity involved, such as ports to listen on and files to create.
-resource development: typing.List[typing.Dict[str, typing.Any]] # Provide an development requirements for each tool, including a Name, Product Goals, Requirement Pool, Anything UNCLEAR. Name: According to the content of "Product Goals," name the project using snake case style , like 'port_scanner'. Product Goals:Provide up to three clear, orthogonal product goals .Requirement Pool:List down the top-5 requirements .Configure: Parameters that need to be configured for product operation. Anything UNCLEAR: Mention unclear project aspects, then try to clarify it, Some background information needed for product development, such as what functions the client and server are responsible for through CS architecture
-resource obtain: typing.List[str] # Specify the resources that need to be downloaded and their sources.
-environmental deployment: typing.List[str] # Outline the deployment steps for each tool on the respective machines.
-Steps In Operation: typing.List[str] # Provide a step-by-step operational procedure on how the tools and environments interact with each other to achieve the goal.
-Verification Checklist: typing.List[str] # List clear and verifiable check items.
## constraint
Format: output wrapped inside [CONTENT][/CONTENT] like format example, nothing else.
resource development: Capabilities that cannot be realized through tools native to the operating system or downloaded from the Internet, and that require our own development. Note that the configuration should be consistent with that defined in "Configure".
resource obtain: Capabilities that can be obtained directly from the operating system or tools downloaded from the Internet
Steps In Operation: Note that the configuration should be consistent with that defined in "Configure".
Verification Checklist: Note that the configuration should be consistent with that defined in "Configure".
## action
Follow instructions of nodes, generate output and make sure it follows the format example.
如下是在Product Architect在完成工具设计和Product Manager完成任务分配后,发送给Developer进行部分代码开发的案例。其中Design是Product Architect基于需求的设计文档,Task是Product Manager基于设计文档制定的项目开发计划,Legacy Code是之前的Agent生成的代码,Instruction是相应的操作指示。
NOTICE
Role: You are a professional develper; the main goal is develop clean, maintainable, and efficient code for the attack scenarios based on the architecture and task assignments.
Constraints: Adhere to coding standards and best practices. Ensure the code is modular, readable, and aligns with the specified design.
ATTENTION: Use '##' to SPLIT SECTIONS, not '#'. Output format carefully referenced "Format example".
# Context
## Design
{"Implementation approach": "We will develop a Python script that interfaces with the Windows Registry API to perform the necessary tasks. We will use the `winreg` module for registry operations and `os` and `sys` modules to check for administrative privileges. The script will include functions to backup and restore the registry, hide user accounts, and log changes.", "File list": ["main.py", "registry_utils.py", "logger.py", "config.py"], "Data structures and interfaces": "\\nclassDiagram\\n class Main {\\n +main() void\\n }\\n class RegistryUtils {\\n +backup_registry() bool\\n +restore_registry(backup_file: str) bool\\n +hide_user_account(username: str) bool\\n +is_admin() bool\\n }\\n class Logger {\\n +log_change(change: str) void\\n }\\n class Config {\\n +backup_path str = \'{registry_backup_path}\'\\n }\\n Main --> RegistryUtils\\n RegistryUtils --> Logger\\n RegistryUtils --> Config\\n", "Program call flow": "\\nsequenceDiagram\\n participant M as Main\\n participant RU as RegistryUtils\\n participant L as Logger\\n participant C as Config\\n M->>RU: is_admin()\\n RU-->>M: return is_admin\\n M->>RU: backup_registry()\\n RU->>C: backup_path\\n C-->>RU: return backup_path\\n RU->>L: log_change(\'Backup created\')\\n L-->>RU: return\\n RU-->>M: return backup_result\\n M->>RU: hide_user_account(username)\\n RU->>L: log_change(\'User account hidden\')\\n L-->>RU: return\\n RU-->>M: return hide_result\\n M->>RU: restore_registry(backup_file)\\n RU->>L: log_change(\'Registry restored\')\\n L-->>RU: return\\n RU-->>M: return restore_result\\n", "Usage": {"server": ["Set \'registry_backup_path\' in config.py", "python main.py"], "client": ["Run as administrator", "python main.py"]}, "Anything UNCLEAR": "Clarification needed on the specific method to determine administrative privileges within the script."}
## Task
{"Required Python packages": ["winreg", "os", "sys"], "Required Other language third-party packages": ["No third-party dependencies required"], "Logic Analysis": [["config.py", "Contains configuration variables such as \'registry_backup_path\'. No dependencies."], ["logger.py", "Contains Logger class to log changes. No dependencies."], ["registry_utils.py", "Contains RegistryUtils class with methods for registry operations. Depends on config.py for configuration variables and logger.py for logging."], ["main.py", "Contains the main function that orchestrates the flow of the program. Depends on registry_utils.py for performing registry operations."]], "Task list": ["config.py", "logger.py", "registry_utils.py", "main.py"], "Full API spec": "", "Shared Knowledge": "`registry_utils.py` contains functions shared across the project for interacting with the Windows Registry.", "Anything UNCLEAR": "Clarification needed on the specific method to determine administrative privileges within the script."}
## Legacy Code
```Code
----- config.py
```## config.py
# This module contains configuration variables for the application.
# It sets default values for necessary configurations and avoids circular imports by not depending on any other custom module.
class Config:
# Default path for registry backup. Can be modified as per requirement.
backup_path: str = "C:\\registry_backup"
def __init__(self):
# Constructor is not necessary for this class as it only contains static configuration data.
# However, it's defined here to follow best practices and for potential future use.
pass
```
----- logger.py
```## logger.py
import datetime
class Logger:
"""
A simple logging class to log changes and events within the application.
Logs are printed with a timestamp.
"""
@staticmethod
def log_change(change: str) -> None:
"""
Logs a change or event with the current timestamp.
Args:
change (str): The description of the change or event to log.
"""
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] {change}")
```
```
## Debug logs
```text
```
## Bug Feedback logs
```text
```
# Format example
## Code: registry_utils.py
```python
## registry_utils.py
...
```
# Instruction: Based on the context, follow "Format example", write code.
## Code: registry_utils.py. Write code with triple quoto, based on the following attentions and context.
1. Only One file: do your best to implement THIS ONLY ONE FILE.
2. COMPLETE CODE: Your code will be part of the entire project, so please implement complete, reliable, reusable code snippets.
3. Set default value: If there is any setting, ALWAYS SET A DEFAULT VALUE, ALWAYS USE STRONG TYPE AND EXPLICIT VARIABLE. AVOID circular import.
4. Follow design: YOU MUST FOLLOW "Data structures and interfaces". DONT CHANGE ANY DESIGN. Do not use public member functions that do not exist in your design.
5. CAREFULLY CHECK THAT YOU DONT MISS ANY NECESSARY CLASS/FUNCTION IN THIS FILE.
6. Before using a external variable/module, make sure you import it first.
7. Write out EVERY CODE DETAIL, DON'T LEAVE TODO.
Agent Pipelines
基于前面定义的不同Role的Agent以及Agent的能力,我们可以把这些Agent按照Pipeline进行编排实现基于分析Procedure Example实现的攻击模拟。
首先由Requirements Analyst对需求进行分析,这一步的核心是明确需要模拟的攻击行为以及定义攻击需要达到的目标。完成需求分析后,会将详细需求发送给Security Architect,由他对攻击的详细步骤进行拆解,并且确定攻击实现对环境和工具的依赖。然后其中需要开发的能力由Product Architect进行产品设计。接着将环境依赖、开发需求、攻击步骤发送给Product Manager,由他进行任务拆分以及任务分配。拆分的任务会分配给Developer进行能力实现,并由SRE部署环境和代码。最后会由Penetration Tester对功能进行测试,同时会将问题反馈给前面的Agent进行修改并继续调试,直至达成攻击目标。
内存管理
Multi-Agent的协作需要记忆机制的参与。通过维护长期的记忆如设计文档、配置信息可以协调不同的Agent共同完成最终的目标。结合执行产出的短时记忆,可以让Agent更好的进行推理。
Static Memory :全局长期不变,主要存储约定的配置、能力等信息,如AWS支持的基础镜像,AWS支持的instance_type 等
Long-Term Memory: 长期记忆,主要维护需要全局使用,持续更新的信息。如Requirements Analyst分析产出的需求和目标信息
Short-Term Memory: 短期记忆,主要维护Agent之间短期传递的信息。如在Penetration Tester在进行攻击模拟时返回的错误信息。
Agent获取太多的内存信息,也会导致Agent的信息的过载。所以需要对内存信息标签化,每一份信息在写入内存时,会附带Tag信息用来指定未来可以获取该信息的Agent。
推理
在一次完整的攻击模拟过程中,推理是必须具备的能力,我们需要基于逻辑或知识通过分析和推断,最终作出决策。BASGPT主要通过ReACT Agent model来实现推理。
如攻击模拟的执行之前,BASGPT会基于前面的分析的攻击目标、设计的攻击过程、开发的攻击能力、部署的攻击环境等信息,最终作出规划,制定整个攻击的具体执行步骤。
而在攻击模拟的执行过程中,避免不了开发问题、部署问题、环境问题导致的异常和错误,我们要基于异常信息作出反思,最终制定相应的迭代动作解决问题。