Всем привет, меня зовут Артём Семенов, я являюсь автором Телеграм канала PWN AI. С момента релиза GPT 3.5 прошло больше 2ух лет. Это явление, я про распространение LLM, стало революционным во всех сферах, включая кибербезопасность. Модели стали применять для разных вещей. Например, использовать как дополнение к SAST, анализатор Vulnhunter от protectai может использоваться для анализа кода веб-приложений на наличие популярных типов уязвимостей – xss, csrf, sql и всё это, не имея за ранее предобученной модели, лишь на основании инструкции происходит анализ кода. Это можно подстроить под концепцию Zero Shot, когда во время работы модель видя новые для неё данные должна сделать предсказание.

Можно также вспомнить примеры, когда llm модели использую как инструмент для генерации данных для фаззинга, а также для анализа принадлежности файлов к чему-либо.

Я уверен вы и сами уже применяете разные модели для того, чтобы упростить себе выполнение задачи. Ну и да - кто я такой чтобы вам говорить о том, что конфиденциальные данные нельзя запихивать в модель). Я думаю, вы это и сами хорошо понимаете.

В какой-то степени – написание статьи это крик моей души. Так как нету ничего на русском что могло бы доступно объяснить специалисту по ИБ, как он может применить с пользой это всё.

Немного о... психологии

Давайте обратимся немного к психологическим аспектам нашего мышления и его эффективного применения в работе. Возьмём концепции медленного и быстрого мышления (об этом детально можно почитать в книге «Думай медленно, решай быстро»).

Быстрое мышление как правило, срабатывает автоматически и зачастую вы не можете контролировать себя, свои эмоции и принять правильное и обоснованное решение. Быстрое мышление выстраивает сложные схемы мыслей, однако формирует краткосрочные прогнозы.

Вот мы можем привести пример того, когда вы подаёте запрос в модель – она быстро отдаёт вам ответ. 90% случаев вы получаете не совсем тот результат, вы делаете несколько дополнений и корректировку запроса. Тем самым вы улучшаете конечный ответ. Хотя не всегда, есть же такое что модель чего-то не знает. Вам нужно дополнить её знания, потом сказать о возникающих ошибках, если это какой-то инструмент и т.д – всё это в конечном счёте приводит к тому что вы тратите много времени. Не обдумав ситуацию, стратегию изначально.

Представим себе если бы при создании чего-либо, при тестировании на безопасность какой-либо системы вы прибегали бы к медленному мышлению. Вы можете уже видеть преимущества в том что вы сможете принять решение на основе изначально огромного количества собранной информации, вы не будете под эмоциональным шквалом и вы сможете в целом продумать как вы будете контролировать свои действия.

Кстати, исключения ради – статья написана именно под первым режимом мышления. Уж очень хотелось показать примеры. Мало зарубежных статьей именно о том, как это можно практически применить.

Как нам реализовать более эффективный режим мышления? Вот тут на помощь и приходят агенты и мультиагентные системы

Если суперкратко, то агент – это компонент системы, принимающий ваш запрос и выполняющий вашу задачу. Агент в свою очередь является частью агентной системы или мультиагентной. В основе лежит ядро – это большая языковая модель, которая отвечает за обработку и принятие решений исходя из цели и задачи, которую вы поставили. Мультиагентной система будет в том случае, если там используется множество агентов, которые в свою очередь могут кооперировать между друг другом и, если необходимо, делегировать друг другу задачи.  

Давайте посмотрим концептуально что необходимо и важно для агента при выполнении задачи. Первое – вы должны определить, чем он вообще будет заниматься, словно вы строите свою дрим-тим, и каждый персонаж имеет определённую роль.

Второе – это то, на чём он должен фокусироваться при выполнении задачи. Мы всегда должны помнить, что от чётко поставленного задания всегда зависит результат, важно, чтобы агент понимал изначально что ему можно, а что нельзя и на чём ему следует фокусироваться.

Третье – инструменты. В большинстве современных фреймворков для создания агентных систем есть возможность использовать внешние инструменты или API, для того чтобы агент мог использовать их при выполнении вашей поставленной задачи. Идеальным примером является Claude Desktop от Antropic, которая может полностью взаимодействовать с вашей операционной системой, используя все инструменты и выполнять различие задачи. Кстати Anthropic недавно выпустили потрясающий гайд о том как проектировать мультиагентные системы – большое спасибо и привет Илье Гусеву за разбор этих принципов. И привет ему.

Четвёртое - кооперация, в основе любой мультиагентной системы, как и команды лежит такое важное свойство как кооперация при выполнении задачи и анализа ошибок. В большинстве фреймворков вы можете реализовать связи между агентами, а также настроить делегирование и асинхронное выполнение задач. Чтобы ваша дрим-тим из железок работала эффективнее над вашими мечтами.

Пятое - память. Агенты, как и люди, могут учиться на своих ошибках, а также иметь краткосрочную и долгосрочную память. Всё это также является важным при создании эффективной системы из агентов.

Друзья, прежде чем мы приступим к созданию такой системы – я хочу дать вам рассмотреть картинку о преимуществах создания софта с агентами.

Во-первых, мы можем видеть различия в том, какие операции с данными нам доступны и их преобразования, во-вторых – это то, что мы можем подать на вход и, в-третьих, это вывод – это крайне обязательно, как от любого сотрудника мы просим чётко что мы хотим получить на выходе – так и тут. Это важно учитывать. Помните, что вы при проектировании такой системы должны действовать как менеджер. Ваше чёткое понимание и определение целей и задач, а также их доступное объяснение для агентов даст большую пользу вам в будущем.

Приступим к практике

Среди множества фреймворков, которые позволяют реализовать мультиагентную систему – мы будем использовать crew.ai.

В чём его преимущества?

Во-первых, простота синтаксиса при создании агентов.

Во-вторых, из коробки идёт множество полезных инструментов, которые можно интегрировать с агентом для того, чтобы упростить выполнение задач

В-третьих, поддержка асинхронного выполнение, памяти, делегации задач, RAG – для того, чтобы черпать информацию из файлов.

В-четвёртых – фреймворк имеет большое сообщество.

В-пятых – это поддержка большого количества моделей, включая ollama и возможность использовать локальные модели через Ollama API.

Я бы мог тут бесконечно продолжать, но всё-таки цель нашей статьи дать вам понимание о том как и зачем применять это всё дело в кибербезе.

Давайте определимся с тем, что будет делать наша мультиагентная система, какой геморрой она будет устранять.

Я предлагаю сделать несколько систем:

• Первая будет анализировать публичные базы уязвимостей и давать рекомендации, на основе описания уязвимости в базе.

• Вторая будет запускать несколько инструментов для того, чтобы решать уязвимое приложение от OWASP.

Начнём с простого. Всё делается через python. Открываем терминал :-) Мы не будем использовать poerty или conda. Хотя вы можете без проблем загнать всё это в виртуальное окружение. Моя рекомендация – использовать uv, так как он быстрее всех.

pip install crewai crewai[tools]

Этой командой мы ставим основные компоненты для создания нашей системы. Первое это сам фреймворк, позволяющий взаимодействовать с ядром, создавать агентов и работать с инструментами. Второе – это сами инструменты. Их уже достаточно много – вот тут вы можете познакомится со всеми инструментами поближе. Некоторые инструменты могут требовать API ключи, так как они предоставляют вам возможность взаимодействия с внешними сервисами, которые могут быть платными.

from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI  # этот импорт нам необходим для того чтобы была llm, которая могла управлять процессами и делегированием в системе.
from crewai_tools import SerperDevTool, \
                         ScrapeWebsiteTool, \
                         WebsiteSearchTool
os.environ["OPENAI_MODEL_NAME"] = 'gpt-4-turbo'
os.environ["OPENAI_API_KEY"] = 'sk-*' # либо можно через загрузку переменных окружений или dotenv(если windows)
os.environ["SERPER_API_KEY"] = ‘serper-key’ # также меняем, получаем его на сайте serper

manager_llm = ChatOpenAI(model="gpt-4")
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()
website_search = WebsiteSearchTool()

В этом блоке кода мы импортируем фреймворк и три его значимых модуля-концепции:

1. Агент(Agent), он будет иметь свою роль, цель, а также подключенный набор инструментов, который ему доступен.

2. Задача(Task), то что будет делать агент или мультиагентная система. В задача важно чётко описать что мы хотим получить на выходе. И чего следует избегать.

3. Crew – тот компонент, который будет распределять задачи между агентами, а также контролировать функционал памяти.

Process определяет то как будут выполнятся задачи, они могут быть выполнены последовательно или может быть реализовано делегирование в зависимости от иерархии.

Дальше мы импортировали crewai_tools, в нём подключены инструменты для отправки запросов в Google – через API Serper(SerprerDevTool), ScrapeWebsiteTool позволяет нам буквально считать всю информацию с сайта(есть также поддержка selenium - SeleniumScrapingTool), а WebsiteSearchTool позволит нам делать обращение к любому сайту через RAG, получая из него информацию.  

Мы их инициализируем в коде ниже, однако они также могут принимать значения – если вы хотите чтобы был поиск по определённым ресурсам то необходимо об этом указать в  функции. Например так:

search_tool = SerperDevTool(
    country="fr",
    locale="fr",
    location="Paris, Paris, Ile-de-France, France",
    n_results=2,
)

или так

search_tool = SerperDevTool(
    search_url="https://google.serper.dev/scholar",
    n_results=2,
)

Для большинства инструментов могут быть применены различные параметры, о каждом можно найти информацию в документации.

Мы также установили важные переменные окружения – например это название модели и API-ключ. Без ядра ничего не заработает. Вы можете указать также claude или другую модель.

Создадим нашего первого агента. Он будет выполнять роль исследователя, который будет смотреть и анализировать базу-уязвимостей на наличие данных, которые мы предоставили.

Первая роль – исследователь, он будет ходить по интернету в поисках информации о CVE.

vulnerability_researcher = Agent(
    role='Vulnerability Researcher',
    goal='Find and analyze new security vulnerabilities with detailed technical information',
    backstory="""You are an expert security researcher with years of experience in 
    vulnerability analysis. Your specialty is finding and analyzing new CVEs and 
    security threats, with a focus on technical details and impact assessment.""",
    verbose=True,
    allow_delegation=True,  # Разрешаем делегирование
    tools=[search_tool, scrape_tool, website_search]
)

Каждый агент описывается отдельно как класс, в нём описывается роль, цель агента, под backstory описывается его инструкция, которая задаёт образ мышления.

Verbose означает вывод информации рассуждении, а также логи.

Allow_delegation – атрибут, указывающий на возможность делегировать задачи.

А tools – буквально описываем инструменты которые можно использовать агенту.

Создадим ещё парочку, так как у нас мультиагентная система.

Следующий агент – аналитик воздействия от уязвимости, он будет оценивать то, насколько она применима к бизнесу и несёт реальный ущерб.

impact_analyzer = Agent(
    role='Impact Analyzer',
    goal='Analyze the business impact and risk level of identified vulnerabilities',
    backstory="""You are a senior security analyst specialized in determining the 
    real-world impact of vulnerabilities. You excel at translating technical 
    vulnerabilities into business risks and providing clear severity ratings.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, website_search]

Следующий агент – эксперт по составлению рекомендаций.

recommendation_expert = Agent(
    role='Security Advisor',
    goal='Provide detailed mitigation strategies and recommendations',
    backstory="""You are a security advisor with extensive experience in 
    vulnerability remediation. You provide practical, actionable recommendations 
    for addressing security vulnerabilities, including specific steps, patches, 
    and best practices.""",
    verbose=True,
    allow_delegation=True,  
    tools=[search_tool, website_search]
)

Дальше, мы создаём задачу. Как правило, задача должна быть подробна описана. Я думал можно описать всё на русском, но, к сожалению, выходит не очень.

research_task = Task(
    description=f"""
    Research and identify new critical vulnerabilities with the following parameters:
    - Vendor: {search_params['vendor']}
    - Product: {search_params['product']}
    - Timeframe: {search_params['timeframe']}
    - Severity: {search_params['severity']}

    Focus on:
    1. Technical details of each vulnerability
    2. Affected systems and versions
    3. Exploitation methods
    4. Current patch status
    
    Format the output as a structured list with clear headers for each vulnerability.
    """,
    agent=vulnerability_researcher,
    expected_output="A detailed list of vulnerabilities with technical details, affected systems, exploitation methods, and patch status"
)

В research_task мы описали задачу через description, также можно добавить атрибут, expected_output - в нём описать что мы хотим видеть по итогу в выводе. В некоторых случаях использую pydantic, для обработки вывода – но в этой статье мы не рассмотрим его. Мы также подаём входные данные, подробное описание будет рассмотрено чуть ниже, перед запуском.

Сама задача буквально заключается в том, чтобы агент нашёл информацию об уязвимости, методы эксплуатации и текущую версию патча. Он буквально должен фокусироваться только на этом.  Для её выполнения будет использоваться агент vulnerability_researcher.

Лучшей практикой является разделение задач. Поэтому создадим ещё 2 задачи.

analysis_task = Task(
    description=f"""
    Analyze the identified vulnerabilities for {search_params['vendor']} {search_params['product']} and determine:
    1. Business impact
    2. Risk level (Critical, High, Medium, Low)
    3. Potential consequences of exploitation
    4. Affected industries
    
    Use the research from the previous task and provide a detailed analysis for each vulnerability.
    """,
    agent=impact_analyzer
)

recommendation_task = Task(
    description=f"""
    Based on the vulnerabilities found for {search_params['vendor']} {search_params['product']}, provide:
    1. Detailed mitigation strategies
    2. Step-by-step remediation instructions
    3. Temporary workarounds where applicable
    4. Long-term security recommendations
    
    Create a comprehensive remediation plan for each vulnerability.
    """,
    agent=recommendation_expert,
    expected_output="A detailed remediation plan including mitigation strategies, step-by-step instructions, workarounds, and long-term recommendations"
)

Я видел примеры, когда всё закидывают под один блок. Но что-то мне подсказывает что это не совсем эффективно, так как может сбиться фокус.

Дальше создадим crew, который буквально соединит наших агентов, расставит на них задачи и позволит нам настроить процесс выполнения и делегирования.

crew = Crew(
    agents=[vulnerability_researcher, impact_analyzer, recommendation_expert],
    tasks=[research_task, analysis_task, recommendation_task],
    verbose=2,
    process=Process.hierarchical,
    manager_llm=manager_llm
)

Важно объявлять crew как Crew – фреймворк иначе не запустится. В нём мы подаём агентов – agents .. задачи – tasks … это всё для того чтобы наша цель была исполнена. Используйте с умом. Если вы в будущем будете делать несколько целей, то может вам не нужно будет указывать все агенты и задачи ?

verbose = 2, тут мы буквально говорим чтобы абсолютно вся информация о принимаемых решениях и то, как происходит делегирование ( не с технической точки зрения, а именно вот как процесс).

manager-llm буквально наследует то, что мы указали при определении переменных окружения и инициализации инструментов.

Рекомендуется также подавать входные данные в формате словаря, где есть значение аргумент, однако вы можете реализовать ввод входных данных из командной строки, если пожелаете.

search_params = {
    "vendor": "Microsoft",  # Производитель
    "product": "Windows",   # Продукт
    "timeframe": "last_week",  # Временной период (last_day, last_week, last_month)
    "severity": "critical"  # Уровень критичности (critical, high, medium, low)
}

Теперь мы укажем в коде функцию для запуска и вывода результатов..

result = crew.kickoff()

# Вывод результатов
print("\n=== Vulnerability Analysis Report ===\n")
print(result) 

Чтобы запустить файл нужно прописать либо:

python3 run_script.py

или

crewai test

Если вы хотите запустить агент в несколько итераций и с другой моделью (важно чтобы вы тогда не указывали в коде другую) то:

crewai test --n_iterations 5 --model gpt-4o

Конечный отчёт выглядит так:

=== Vulnerability Analysis Report ===

Remediation Plan for CVE-2024-49138: Heap-Based Buffer Overflow Vulnerability in Microsoft Windows CLFS Driver:

1. Immediate Actions:
   - Apply the Patch: Install the latest security updates provided by Microsoft. You can find the necessary updates in the Microsoft Security Update Guide specific to CVE-2024-49138. Follow the links from the Search Results for direct access.
   - Mitigation Script: Utilize the mitigation script provided by sources like Vicarius, which specifically adjusts CLFS Service Permissions. This script is to limit the CLFS service's permissions till the patch can be applied.

2. Monitoring and Validation:
   - Monitor system logs and security systems for any anomalous activities that suggest exploitation attempts.
   - Validate that the patch and mitigation measures have been successfully implemented using vulnerability scanning tools and compliance checks.

3. Long-Term Security Recommendations:
   - Regularly update all systems with the latest patches released by Microsoft.
   - Implement a robust patch management program to ensure the timely application of security updates.
   - Enhance monitoring on systems handling sensitive data or critical operations.

Remediation Plan for CVE-2024-49112: Vulnerability in Lightweight Directory Access Protocol (LDAP):

1. Immediate Actions:
   - Apply the Patch: As with the previous vulnerability, immediately install the latest security updates made available by Microsoft for the LDAP protocol vulnerabilities under CVE-2024-49112. Reference the Microsoft Security Update Guide for specific patches.
   - Network Controls: Implement network segmentation and access controls to minimize the potential impact of an exploit. Restrict LDAP traffic to necessary systems and block unnecessary external access points.

2. Monitoring and Validation:
   - Continuous monitoring for exploitation attempts against LDAP services. Pay special attention to any unexpected external connections or unauthorized access attempts.
   - Conduct a post-patch validation to ensure the patch has been successfully applied and the vulnerability is mitigated.

3. Long-Term Security Recommendations:
   - Regularly review and update LDAP configurations and security settings. Ensure that LDAP servers are hardened against attacks and abide by the principle of least privilege.
   - Undertake security training for IT staff to recognize the symptoms of exploitation attempts and respond accordingly.

Further Steps:
   - Conduct regular security audits and penetration tests to evaluate the effectiveness of the implemented security measures.
   - Stay informed about new threats by following the Microsoft Security Response Center and other cybersecurity publications.

By following these detailed steps, your organization will not only mitigate the immediate risks associated with CVE-2024-49138 and CVE-2024-49112 but also enhance your overall security posture against future vulnerabilities.

Хоть и кажется, что более общим языком. Но помните – если вы менеджер, вы должны детально описать что для вас «Глубина» и «Вода» в тексте, а не нести абстрактные понятия и думать, что задача будет великолепно исполнена. Особенно когда мы работаем с анализом. Продумайте роли детально. 

Можно использовать асинхронный запуск, если пожелаете. Оно будет особенно применим, если вы запускаете несколько мультиагентных систем одновременно.

Часть 2. Делаем систему которая будет ломать систему.

Хоть мы и не xbow.com, но было бы прикольно сделать базовый решатор тачек типа dvwa, который будет искать уязвимости, репортить их. И ясное дело что эта штука не заменит пентестеров. Ну, потому что кейсы в большинстве случаев уникальны. Хотя xbow стремится доказать, что можно заменить хакеров. Не верю.

Что нам нужно в этом случае? Давайте начнём с импортов:

import os
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
from crewai_tools import SerperDevTool, ScrapeWebsiteTool, WebsiteSearchTool
from crewai.tools import BaseTool
import requests
from typing import Optional
from pydantic import BaseModel, Field
import re
import time
from concurrent.futures import ThreadPoolExecutor, TimeoutError

# Настройка переменных окружения
os.environ["OPENAI_MODEL_NAME"] = 'gpt-4-turbo'
os.environ["OPENAI_API_KEY"] = 'sk'
os.environ["SERPER_API_KEY"] = 'b1'

практически всё также, мы добавили инструменты для работы с данными, регулярными выражениями и потоки … почему не асинхронные агенты ? Да того чтобы система делала иногда остановку…

Самое сложное – это реализовать инструмент для взаимодействия с DVWA. Таких вот инструментов ну по факту нету из коробки. Их нужно сделать самостоятельно. Важно понимать стек решения. Допустим у dvwa авторизация по csrf, payloads оно принимает определённым образом в url. Да и мы помним что авторизационный endpoint – это login.php

Нам нужен отдельный DVWAtool !

class DVWAInteractionTool(BaseTool):
    name: str = "DVWAInteractionTool"
    description: str = """
    Инструмент для взаимодействия с DVWA (Damn Vulnerable Web Application).
    Формат команды: 'test|vulnerability_type|payload|url'
    Например: 'test|sqli|1' OR '1'='1|http://localhost/vulnerabilities/sqli/'
    """
    session: requests.Session = Field(default_factory=requests.Session)
    base_url: str = Field(default="http://localhost")
    logged_in: bool = Field(default=False)

    class Config:
        arbitrary_types_allowed = True

    def login(self) -> bool:
        """Авторизация в DVWA"""
        try:
            # Сначала пробуем войти без CSRF токена
            login_data = {
                'username': 'admin',
                'password': 'password',
                'Login': 'submit'
            }

            # Отправляем запрос на вход
            login_url = f"{self.base_url}/login.php"
            response = self.session.post(login_url, data=login_data, allow_redirects=True)
            print(f"Login response status: {response.status_code}")
            print(f"Login response headers: {response.headers}")
            print(f"Login response text: {response.text[:200]}")

            # Проверяем успешность входа
            if response.status_code == 200:
                if 'index.php' in response.text or 'Welcome' in response.text:
                    self.logged_in = True
                    return True
                
                # Если нужен CSRF токен, пробуем его найти
                csrf_match = re.search(r'input\s+type=["\']hidden["\']\s+name=["\']user_token["\']\s+value=["\'](.*?)["\']', response.text, re.I)
                if csrf_match:
                    csrf_token = csrf_match.group(1)
                    print(f"Found CSRF token: {csrf_token}")
                    
                    # Пробуем войти с CSRF токеном
                    login_data['user_token'] = csrf_token
                    response = self.session.post(login_url, data=login_data, allow_redirects=True)
                    self.logged_in = 'index.php' in response.text or 'Welcome' in response.text
                    return self.logged_in

            return False

        except Exception as e:
            print(f"Login error: {str(e)}")
            return False

    def _extract_csrf_token(self, html: str) -> str:
        """Извлечение CSRF токена из HTML"""
        import re
        # Обновите регулярное выражение в соответствии с HTML-кодом страницы
        match = re.search(r'name="user_token" value="([^"]+)"', html)
        if match:
            return match.group(1)
        else:
            print("CSRF token not found in HTML.")
            return ''

    def test_vulnerability(self, vuln_type: str, payload: str, url: str) -> str:
        """Тестирование конкретной уязвимости"""
        if not self.logged_in:
            if not self.login():
                return "Failed to login to DVWA"

        try:
            # Добавляем таймаут для запросов
            timeout = 10  # секунд
            
            if vuln_type == "sqli":
                return self._test_sqli(payload, url, timeout)
            elif vuln_type == "xss":
                return self._test_xss(payload, url, timeout)
            else:
                return f"Unsupported vulnerability type: {vuln_type}"

        except Exception as e:
            return f"Error testing {vuln_type}: {str(e)}"

    def _test_sqli(self, payload: str, url: str, timeout: int) -> str:
        """Тестирование SQL-инъекции"""
        params = {"id": payload, "Submit": "Submit"}
        response = self.session.get(url, params=params, timeout=timeout)
        return self._analyze_response(response, "sqli", payload)

    def _test_xss(self, payload: str, url: str, timeout: int) -> str:
        """Тестирование XSS"""
        params = {"name": payload, "Submit": "Submit"}
        response = self.session.get(url, params=params, timeout=timeout)
        return self._analyze_response(response, "xss", payload)

    def _analyze_response(self, response: requests.Response, vuln_type: str, payload: str = "") -> str:
        """Анализ ответа от приложения"""
        if response.status_code != 200:
            return f"Error: HTTP {response.status_code}"

        print(f"Analyzing response for {vuln_type}")
        print(f"Response content: {response.text[:200]}")  # Первые 200 символов для отладки

        # Анализ ответа в зависимости от типа уязвимости
        if vuln_type == "sqli":
            if "User ID exists" in response.text or "mysql" in response.text.lower():
                return f"SQL Injection potentially successful with payload: {payload}"
            return f"SQL Injection attempt failed with payload: {payload}"
        
        elif vuln_type == "xss":
            if payload in response.text:
                return "XSS successful: Payload reflected in response"
            return "XSS attempt: Payload not reflected"
        
        elif vuln_type == "command":
            if "ping -c 4" in response.text:
                return "Command Injection possible: Command output visible"
            return "Command Injection attempt: No clear indication of success"
        
        elif vuln_type == "file_inclusion":
            if "Failed opening" not in response.text:
                return "File Inclusion successful: File content retrieved"
            return "File Inclusion attempt: File not accessible"

        return f"Response received for {vuln_type}: {len(response.text)} bytes"

    def _run(self, command: str) -> str:
        """Обработка команды от агента"""
        try:
            parts = command.split('|')
            if len(parts) != 4:
                return "Invalid command format. Use: test|vulnerability_type|payload|url"

            action, vuln_type, payload, url = parts
            if action != "test":
                return f"Unknown action: {action}"

            return self.test_vulnerability(vuln_type, payload, url)

        except Exception as e:
            return f"Error processing command: {str(e)}"

Да, не идеально. Буквально вот его код, это отдельный класс. Он содержит в себе проверку url, авторизацию по csrf, и описание какие уязвимости нужно проверить, а также регулярки … для поиска csrf и проверки что всё успешно.

А также есть таймауты. Зачем? – потому что иногда система может перестать реагировать вовсе. Наш crew.kickoff, отвечающий за запуск будет исполнятся с таймаутом:

def run_with_timeout(crew, timeout=300):  # 5 минут таймаут
    with ThreadPoolExecutor() as executor:
        future = executor.submit(crew.kickoff)
        try:
            result = future.result(timeout=timeout)
            return result
        except TimeoutError:
            return "Process timed out after {} seconds".format(timeout)

# Запуск процесса с таймаутом
result = run_with_timeout(crew)

До того, как мы определим функцию с таймаутами – мы делаем инициализацию всех инструментов:

# Инициализация инструментов
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()
website_search = WebsiteSearchTool()
dvwa_tool = DVWAInteractionTool()

# Настройка LLM для менеджера
manager_llm = ChatOpenAI(model="gpt-4")

описываем роли и агентов как раньше, а также даём им задачи

# Создание агентов
vulnerability_researcher = Agent(
    role='DVWA Vulnerability Researcher',
    goal='Identify and analyze security vulnerabilities in DVWA application',
    backstory="""You are an expert security researcher specializing in web application security.
    Your focus is on identifying OWASP Top 10 vulnerabilities and providing detailed technical analysis.
    You have extensive experience with common web vulnerabilities like SQL Injection, XSS, and CSRF.""",
    verbose=True,
    allow_delegation=True,
    tools=[search_tool, scrape_tool, website_search, dvwa_tool]
)

impact_analyzer = Agent(
    role='Security Impact Analyzer',
    goal='Analyze the severity and impact of identified vulnerabilities in DVWA',
    backstory="""You are a senior security analyst who excels at risk assessment.
    You understand both technical and business implications of security vulnerabilities.
    You can prioritize vulnerabilities based on their potential impact and exploitation difficulty.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, website_search, dvwa_tool]
)

remediation_expert = Agent(
    role='Security Remediation Expert',
    goal='Provide detailed remediation strategies for DVWA vulnerabilities',
    backstory="""You are a security remediation specialist with deep knowledge of secure coding practices.
    You excel at providing practical, actionable fixes for web application vulnerabilities.
    You understand OWASP security controls and best practices for web application security.""",
    verbose=True,
    allow_delegation=True,
    tools=[search_tool, website_search, dvwa_tool]
)

# Создание задач
research_task = Task(
    description="""
    Investigate DVWA application for SQL Injection and XSS vulnerabilities.
    Focus on:
    1. SQL Injection:
       - Test endpoint: http://localhost/vulnerabilities/sqli/
       - Try payload: 1' OR '1'='1
    2. Cross-Site Scripting (XSS):
       - Test endpoint: http://localhost/vulnerabilities/xss_r/
       - Try payload: <script>alert('XSS')</script>
    
    For each vulnerability:
    - Document the vulnerability type
    - Provide proof of concept
    - Determine if the attack was successful
    
    Format findings as a structured report.
    """,
    agent=vulnerability_researcher,
    expected_output="Technical report of SQL Injection and XSS vulnerability testing results"
)

analysis_task = Task(
    description="""
    Analyze the identified SQL Injection and XSS vulnerabilities:
    1. Assess severity (Critical, High, Medium, Low)
    2. Determine potential impact
    3. Evaluate exploitation complexity
    
    Focus on immediate security risks and potential damage.
    """,
    agent=impact_analyzer,
    expected_output="Risk assessment of discovered SQL Injection and XSS vulnerabilities"
)

remediation_task = Task(
    description="""
    Provide remediation steps for SQL Injection and XSS vulnerabilities:
    1. Input validation recommendations
    2. Output encoding requirements
    3. Security headers configuration
    
    Focus on practical, immediate fixes.
    """,
    agent=remediation_expert,
    expected_output="Specific remediation steps for SQL Injection and XSS vulnerabilities"
)

# Создание crew с последовательным процессом
crew = Crew(
    agents=[vulnerability_researcher, impact_analyzer, remediation_expert],
    tasks=[research_task, analysis_task, remediation_task],
    verbose=True,
    process=Process.sequential  
)

result = run_with_timeout(crew) # позволит нам запустить нашу систему.

Видим, что тестирование удалось и есть рекомендации. Но этот способ имеет проблемы – во-первых, мы опять же видим инструкцию, которая ну не совсем детальные рекомендации имеет. Мы должны с вами сказать об этом, для этого в функцию можно добавить human input execution. В remediation task:

remediation_task = Task(
    description="""
    Provide remediation steps for SQL Injection and XSS vulnerabilities:
    1. Input validation recommendations
    2. Output encoding requirements
    3. Security headers configuration

    Focus on practical, immediate fixes.
    """,
    agent=remediation_expert,
    human_input=True,
    expected_output="Specific remediation steps for SQL Injection and XSS vulnerabilities"
)

В данном случае ввод будет применён только к агенту, который даёт рекомендации.

А и кстати, вы можете спросить – а зачем RAG вообще, так вот – тут буквально происходит подгрузка рекомендаций из интернета в chromadb, эта db используется в фреймворке во всех функциях, которые взаимодействуют с RAG.

Во вторых – лучше запуск таких вот инструментов делать асинхронным.

import asyncio 
...

async def run_crew_async():
    try:
        # Запускаем crew асинхронно
        result = await crew.kickoff_async()
        print("\n=== DVWA Vulnerability Assessment Report ===\n")
        print(result)
        
        # Сохраняем результат в файл
        with open("dvwa_security_report.md", "w") as f:
            f.write("# DVWA Security Assessment Report\n\n")
            f.write(result)
            
        return result
    except Exception as e:
        print(f"Error during crew execution: {str(e)}")
        return f"Error: {str(e)}"

# Запуск асинхронного процесса
if __name__ == "__main__":
    result = asyncio.run(run_crew_async())

Да и следует избирательно подходить к определению таймаута в таких вещах.

Заключение

В этой статье мы разобрали с вами важную концепцию для индустрии – использование агентов. Я уверен в том, что это буквально то будущее, которое нам уже доступно. Хоть и на простых примерах, но мы смогли реализовать то что дало большой импакт и позволило нам упросить ряд наших рабочих действий.

Вы всегда должны помнить о том, что это не совсем безопасно. Некоторые фреймворки предоставляют возможность запуска кода, что может создать риски – если вы не понимаете систему. А иногда агенты могут потерять контроль. Однако, это всё решаемо. Создают агентов которые проверяют и валидируют ввод, и в целом некоторые вендоры начинают задумываться о внедрении функционала безопасности от базовых угроз типа промпт инъекций. Сейчас OWASP пытается реализовать практики безопасного кодирования агентных систем – вы можете об этом получать информацию в моём канале, если вам интересно. Я стараюсь там писать важные новости и писать о теме безопасности ИИ.  

В одном закрытом чатике мы обсуждали применение агентов для дополнение к SAST, так как это бы позволило бы оценивать false positive, а также использовать дополнительные знания помимо правил для определения уязвимостей. SOC-аналитик может использовать это для обработки аллертов. Да и вообще, тут много применений может быть – всё зависит от вашей фантазии.