Автор темы
Разбор мультиагентной системы Swarm Nexus: Автоматическая генерация и SecOps-валидация Windows ctypes-скриптов
Привет, Codeby!
Представляю архитектурный разбор конструктора-оркестратора на Python, разработанного для автоматического синтеза, декомпозиции и безопасного ревью низкоуровневого ИБ-инструментария (WinAPI/ctypes) с использованием локальных LLM (Qwen 2.5 Coder / Heretic).
Архитектурная схема пайплайна
Работа оркестратора разделена на 4 изолированных этапа, управляемых выделенными агентами через Ollama:
[Идея / Задача]
│
▼
[ Agent: Analyst ] ──► Выделение критериев и Regex-контрактов
│
▼
[ Agent: Coder ] ──► Выбор шаблона (Token/Registry/Proc/File) -> Синтез кода
│
┌────►▼
│ [ ОТК / Подложка ]
│ ├── AST-анализ (Запрет встроенного open(), детекция placeholder-заглушек)
│ ├── Поиск утечек хэндлов (OpenProcessToken, CreateFileW, FindFirstFileW)
│ └── Динамический тест (Windows Sandbox / Subprocess с таймаутом)
│ │
│ [Результат теста]
│ ├── Ошибка? ──► [ Agent: Refactor (Refactor Fallback: DeepSeek-R1) ] ──┘
│ └── Успех
│
▼
[ Agent: SecOps ] ──► Финальное ревью артефакта на галлюцинации и критерии безопасности
│
▼
[ Конечный артефакт в /output/ ]
Ключевые технические фичи реализации
1. Автоматический патчинг багов ИИ при работе с ctypes
Локальные модели часто ошибаются в структурах и выравнивании Windows-типов. В коде оркестратора реализован слой _apply_all_fixes, который на лету исправляет артефакты перед отправкой в песочницу:
2. AST-фильтрация и динамический Sandbox-тест
Код запускается внутри изолирующей среды Windows Job Objects. Перед этим абстрактное синтаксическое дерево (AST) сканируется: любые попытки использовать нативный Python-вызов open() вместо WinAPI для работы с файлами, либо обнаруженные ИИ-заглушки вида pass или NotImplementedError расцениваются как саботаж, отправляя код на рефакторинг.
LLM-оркестрация против Access Violation: автоматическая генерация и SecOps-валидация Windows ctypes-скриптов
При написании инструментов автоматизации или малвари на Python под Windows неизбежно приходится слезать с безопасного высокоуровневого рантайма и прыгать в кроличью нору WinAPI через ctypes. И здесь каждого разработчика ждут классические грабли: Access Violation из-за неверного выравнивания полей в структурах, падения из-за некорректной типизации аргументов (argtypes/restype), синтаксические ловушки сырых строк и банальные утечки нативных дескрипторов (Handles).
В этой статье я покажу архитектуру комплексного решения, которое автоматизирует этот процесс. Мы объединим мультиагентный LLM-рой для генерации/рефакторинга кода и изолированную низкоуровневую песочницу на базе Windows Job Objects для безопасного runtime-тестирования сгенерированных скриптов.
Архитектура системы: Мультиагентный рой (Swarm Nexus)
Простая отправка промпта в одну модель часто выдает нерабочий или небезопасный WinAPI-код. Для решения этой проблемы развертывается конвейер из специализированных агентов, которые контролируют друг друга:
Безопасный рантайм: Изоляция через Windows Job Objects
Давать LLM генерировать и сразу запускать WinAPI-скрипты в системе — это прямая дорога к дестабилизации ОС или выполнению деструктивного кода. Обычный subprocess.Popen не защитит систему, если сгенерированный скрипт породит дочерние процессы или зависнет в бесконечном CPU-цикле.
Для изоляции используется механизм Job Objects. Наш модуль песочницы запускает процесс в приостановленном состоянии (CREATE_SUSPENDED), жестко привязывает его дескриптор к объекту задания Windows, выставляет лимиты по CPU-времени (JOB_OBJECT_LIMIT_JOB_TIME), включает принудительное уничтожение дочерних веток (JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE) и только после этого возобновляет поток.
Реализация подсистемы изоляции:
Практический кейс: Перечисление системных файлов через FindFirstFileW
Давайте протестируем систему боевой задачей. Передаем оркестратору ТЗ на низкоуровневый обход файлов в системной директории:
:
Заключение
Использование связки из LLM-роя и низкоуровневых Job Objects позволяет решить две важнейшие задачи:
Итоги бета-теста локальных ИИ
Эксперимент доказывает: внедрение жестких шаблонов проектирования, нативной низкоуровневой подложки рантайм-валидации (WindowsSandbox) и слоев автоматического исправления синтаксиса позволяет добиться стабильной генерации системного софта от локальных моделей класса 7B–14B. Оркестратор успешно «докручивает» галлюцинации ИИ до полностью рабочего состояния, отсекая утечки ресурсов и падения в Access Violation.
Да, это узкоспециализированный конструктор, требующий калибровки под конкретные задачи и лимиты памяти. Тем не менее, как полигон для обкатки мультиагентных систем в автоматизации ИБ — это полностью прозрачный и рабочий стек.[/CODE][/SPOILER]
Привет, Codeby!
Представляю архитектурный разбор конструктора-оркестратора на Python, разработанного для автоматического синтеза, декомпозиции и безопасного ревью низкоуровневого ИБ-инструментария (WinAPI/ctypes) с использованием локальных LLM (Qwen 2.5 Coder / Heretic).
Система полностью автоматизирует цикл от анализа сырой идеи до генерации кода, динамического тестирования артефакта в изолированной подложке и прохождения финального SecOps-контроля.Дисклеймер: Проект не является законченным коммерческим решением. Это гибкий экспериментальный конструктор-оркестратор, собиравшийся «под себя» (под конфигурацию и лимиты конкретного локального железа). Это наглядное пособие и бета-тест для понимания реальных границ возможностей локальных ИИ на данный момент развития технологий.
Архитектурная схема пайплайна
Работа оркестратора разделена на 4 изолированных этапа, управляемых выделенными агентами через Ollama:
- Аналитик (analyst): Извлекает из описания задачи ключевые WinAPI-функции, формирует «критерии проверки» (AST-маркеры) и «контракт вывода» (Regex-паттерны для stdout).
- Кодер (coder): На основе ТЗ и строго типизированных безопасных шаблонов (Registry, Tokens, Process, File, Find) генерирует ctypes-код на Python.
- ОТК/Ревьюер (refactor): Итеративно отлавливает синтаксические ошибки, утечки дескрипторов (Handles), Access Violation и логические несоответствия контракту вывода. При зацикливании выполняет fallback на тяжелую рассуждающую модель (deepseek-r1:14b).
- Валидатор (secops): Финальный гейт качества. Проводит кросс-проверку соответствия сгенерированного артефакта исходным требованиям.
[Идея / Задача]
│
▼
[ Agent: Analyst ] ──► Выделение критериев и Regex-контрактов
│
▼
[ Agent: Coder ] ──► Выбор шаблона (Token/Registry/Proc/File) -> Синтез кода
│
┌────►▼
│ [ ОТК / Подложка ]
│ ├── AST-анализ (Запрет встроенного open(), детекция placeholder-заглушек)
│ ├── Поиск утечек хэндлов (OpenProcessToken, CreateFileW, FindFirstFileW)
│ └── Динамический тест (Windows Sandbox / Subprocess с таймаутом)
│ │
│ [Результат теста]
│ ├── Ошибка? ──► [ Agent: Refactor (Refactor Fallback: DeepSeek-R1) ] ──┘
│ └── Успех
│
▼
[ Agent: SecOps ] ──► Финальное ревью артефакта на галлюцинации и критерии безопасности
│
▼
[ Конечный артефакт в /output/ ]
Ключевые технические фичи реализации
1. Автоматический патчинг багов ИИ при работе с ctypes
Локальные модели часто ошибаются в структурах и выравнивании Windows-типов. В коде оркестратора реализован слой _apply_all_fixes, который на лету исправляет артефакты перед отправкой в песочницу:
- Проблема выравнивания: Принудительное инжектирование pack = 4 для структур WIN32_FIND_DATAW и FILETIME, что полностью решает проблему падения интерпретатора в Access Violation writing.
- Контроль утечек ресурсов: Статический анализ пар вызовов OpenProcessToken -> CloseHandle, CreateFileW -> CloseHandle, FindFirstFileW -> FindClose. Если хэндл не закрыт — скрипт завернется обратно.
- Коррекция WinAPI-логики: Перехват попыток вызова AdjustTokenPrivileges (включение прав), если в исходном таске требовалась только проверка прав (GetTokenInformation).
2. AST-фильтрация и динамический Sandbox-тест
Код запускается внутри изолирующей среды Windows Job Objects. Перед этим абстрактное синтаксическое дерево (AST) сканируется: любые попытки использовать нативный Python-вызов open() вместо WinAPI для работы с файлами, либо обнаруженные ИИ-заглушки вида pass или NotImplementedError расцениваются как саботаж, отправляя код на рефакторинг.
LLM-оркестрация против Access Violation: автоматическая генерация и SecOps-валидация Windows ctypes-скриптов
При написании инструментов автоматизации или малвари на Python под Windows неизбежно приходится слезать с безопасного высокоуровневого рантайма и прыгать в кроличью нору WinAPI через ctypes. И здесь каждого разработчика ждут классические грабли: Access Violation из-за неверного выравнивания полей в структурах, падения из-за некорректной типизации аргументов (argtypes/restype), синтаксические ловушки сырых строк и банальные утечки нативных дескрипторов (Handles).
В этой статье я покажу архитектуру комплексного решения, которое автоматизирует этот процесс. Мы объединим мультиагентный LLM-рой для генерации/рефакторинга кода и изолированную низкоуровневую песочницу на базе Windows Job Objects для безопасного runtime-тестирования сгенерированных скриптов.
Архитектура системы: Мультиагентный рой (Swarm Nexus)
Простая отправка промпта в одну модель часто выдает нерабочий или небезопасный WinAPI-код. Для решения этой проблемы развертывается конвейер из специализированных агентов, которые контролируют друг друга:
- Analyst (Аналитик): Принимает бизнес-задачу, проектирует строгое техническое задание, вычленяет необходимые для проверки WinAPI-функции и формирует жесткий контракт вывода (Output Contract) на базе регулярных выражений.
- Coder (Кодер): Специализированная модель, обученная правилам ctypes. Заполняет шаблон кода, настраивает сигнатуры функций, объявляет C-структуры Windows.
- Refactor (Рефактор): Подключается в случае падения скрипта на этапе runtime-тестирования. Анализирует Traceback и точечно устраняет WinAPI-специфичные баги.
- SecOps (Валидатор): Проводит статический аудит финального кода на полноту реализации контракта безопасности и отсутствие скрытых уязвимостей.
не влезает весь скрипт ниже в прикреплённых файлах переименуй в swarm_nexus_2026.py
Безопасный рантайм: Изоляция через Windows Job Objects
Давать LLM генерировать и сразу запускать WinAPI-скрипты в системе — это прямая дорога к дестабилизации ОС или выполнению деструктивного кода. Обычный subprocess.Popen не защитит систему, если сгенерированный скрипт породит дочерние процессы или зависнет в бесконечном CPU-цикле.
Для изоляции используется механизм Job Objects. Наш модуль песочницы запускает процесс в приостановленном состоянии (CREATE_SUSPENDED), жестко привязывает его дескриптор к объекту задания Windows, выставляет лимиты по CPU-времени (JOB_OBJECT_LIMIT_JOB_TIME), включает принудительное уничтожение дочерних веток (JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE) и только после этого возобновляет поток.
Реализация подсистемы изоляции:
Для проверки надежности песочницы (корректность перехвата таймаутов, дедлоков и жесткого убийства дочерних процессов, переживших родителя) написан следующий набор юнит-тестов:Python:import ctypes from ctypes import wintypes import subprocess import sys # Строгое ограничение: модуль предназначен исключительно для Windows if sys.platform != "win32": raise ImportError("Модуль windows_sandbox предназначен только для Windows.") # --- Константы WinAPI --- JOBOBJECT_EXTENDED_LIMIT_INFORMATION = 9 JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x00002000 JOB_OBJECT_LIMIT_JOB_TIME = 0x00000004 # Лимит CPU-времени ВСЕГО дерева процессов # Права доступа PROCESS_SET_QUOTA = 0x0100 PROCESS_TERMINATE = 0x0001 THREAD_SUSPEND_RESUME = 0x0002 # Флаги CreateProcess / Popen CREATE_SUSPENDED = 0x00000004 CREATE_NO_WINDOW = 0x08000000 # CreateToolhelp32Snapshot TH32CS_SNAPTHREAD = 0x00000004 ERROR_ACCESS_DENIED = 5 ERROR_NO_MORE_FILES = 18 class JOBOBJECT_BASIC_LIMIT_INFORMATION_STRUCT(ctypes.Structure): _fields_ = [ ("PerProcessUserTimeLimit", wintypes.LARGE_INTEGER), ("PerJobUserTimeLimit", wintypes.LARGE_INTEGER), ("LimitFlags", wintypes.DWORD), ("MinimumWorkingSetSize", ctypes.c_size_t), ("MaximumWorkingSetSize", ctypes.c_size_t), ("ActiveProcessLimit", wintypes.DWORD), ("Affinity", ctypes.c_size_t), ("PriorityClass", wintypes.DWORD), ("SchedulingClass", wintypes.DWORD), ] class IO_COUNTERS(ctypes.Structure): _fields_ = [ ("ReadOperationCount", ctypes.c_ulonglong), ("WriteOperationCount", ctypes.c_ulonglong), ("DataOperationCount", ctypes.c_ulonglong), ("ReadTransferCount", ctypes.c_ulonglong), ("WriteTransferCount", ctypes.c_ulonglong), ("DataTransferCount", ctypes.c_ulonglong), ] class JOBOBJECT_EXTENDED_LIMIT_INFORMATION_STRUCT(ctypes.Structure): _fields_ = [ ("BasicLimitInformation", JOBOBJECT_BASIC_LIMIT_INFORMATION_STRUCT), ("IoInfo", IO_COUNTERS), ("ProcessMemoryLimit", ctypes.c_size_t), ("JobMemoryLimit", ctypes.c_size_t), ("PeakProcessMemoryLimit", ctypes.c_size_t), ("PeakJobMemoryLimit", ctypes.c_size_t), ] class THREADENTRY32(ctypes.Structure): _fields_ = [ ("dwSize", wintypes.DWORD), ("cntUsage", wintypes.DWORD), ("th32ThreadID", wintypes.DWORD), ("th32OwnerProcessID", wintypes.DWORD), ("tpBasePri", wintypes.LONG), ("tpDeltaPri", wintypes.LONG), ("dwFlags", wintypes.DWORD), ] # --- Прототипы kernel32 --- kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) kernel32.CreateJobObjectW.argtypes = [ctypes.c_void_p, wintypes.LPCWSTR] kernel32.CreateJobObjectW.restype = wintypes.HANDLE kernel32.SetInformationJobObject.argtypes = [ wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD, ] kernel32.SetInformationJobObject.restype = wintypes.BOOL kernel32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] kernel32.OpenProcess.restype = wintypes.HANDLE kernel32.OpenThread.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] kernel32.OpenThread.restype = wintypes.HANDLE kernel32.AssignProcessToJobObject.argtypes = [wintypes.HANDLE, wintypes.HANDLE] kernel32.AssignProcessToJobObject.restype = wintypes.BOOL kernel32.TerminateJobObject.argtypes = [wintypes.HANDLE, wintypes.UINT] kernel32.TerminateJobObject.restype = wintypes.BOOL kernel32.ResumeThread.argtypes = [wintypes.HANDLE] kernel32.ResumeThread.restype = wintypes.DWORD # (DWORD)-1 при ошибке kernel32.CloseHandle.argtypes = [wintypes.HANDLE] kernel32.CloseHandle.restype = wintypes.BOOL kernel32.CreateToolhelp32Snapshot.argtypes = [wintypes.DWORD, wintypes.DWORD] kernel32.CreateToolhelp32Snapshot.restype = wintypes.HANDLE kernel32.Thread32First.argtypes = [wintypes.HANDLE, ctypes.POINTER(THREADENTRY32)] kernel32.Thread32First.restype = wintypes.BOOL kernel32.Thread32Next.argtypes = [wintypes.HANDLE, ctypes.POINTER(THREADENTRY32)] kernel32.Thread32Next.restype = wintypes.BOOL _INVALID_HANDLE_VALUE = wintypes.HANDLE(-1).value def _find_main_thread_id(pid: int) -> int: """ Возвращает TID главного потока процесса pid через снимок потоков системы. Главный поток — обычно поток с минимальным TID среди потоков процесса. """ snap = kernel32.CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) if snap == _INVALID_HANDLE_VALUE or not snap: raise OSError( f"CreateToolhelp32Snapshot failed. " f"Windows Error Code: {ctypes.get_last_error()}" ) try: entry = THREADENTRY32() entry.dwSize = ctypes.sizeof(THREADENTRY32) if not kernel32.Thread32First(snap, ctypes.byref(entry)): raise OSError( f"Thread32First failed. " f"Windows Error Code: {ctypes.get_last_error()}" ) best_tid = 0 while True: if entry.th32OwnerProcessID == pid: if best_tid == 0 or entry.th32ThreadID < best_tid: best_tid = entry.th32ThreadID if not kernel32.Thread32Next(snap, ctypes.byref(entry)): err = ctypes.get_last_error() if err != ERROR_NO_MORE_FILES: raise OSError( f"Thread32Next failed. Windows Error Code: {err}" ) break if best_tid == 0: raise OSError(f"Не найден главный поток процесса PID {pid}.") return best_tid finally: kernel32.CloseHandle(snap) class WindowsSandbox: """Изолирующее окружение для рантайм-тестов недоверенного кода через Job Objects. Модель изоляции: * Процесс запускается в CREATE_SUSPENDED, привязывается к Job Object, только затем возобновляется через ResumeThread -> окно гонки отсутствует. * JOB_OBJECT_LIMIT_JOB_TIME — лимит суммарного CPU-времени дерева процессов. Wall-clock ограничивается отдельно через Popen.communicate(timeout=...). * JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE — при любом выходе из контекста все процессы дерева уничтожаются, включая потомков. """ def __init__(self, time_limit_ms: int = 5000): if not (0 < time_limit_ms <= 86_400_000): raise ValueError("time_limit_ms должен быть в диапазоне 1..86400000 (24 ч).") self.job_handle = None self.time_limit_ms = time_limit_ms def __enter__(self): self.job_handle = kernel32.CreateJobObjectW(None, None) if not self.job_handle: raise OSError( f"CreateJobObjectW failed. " f"Windows Error Code: {ctypes.get_last_error()}" ) limits = JOBOBJECT_EXTENDED_LIMIT_INFORMATION_STRUCT() limits.BasicLimitInformation.LimitFlags = ( JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE | JOB_OBJECT_LIMIT_JOB_TIME ) limits.BasicLimitInformation.PerJobUserTimeLimit = self.time_limit_ms * 10000 if not kernel32.SetInformationJobObject( self.job_handle, JOBOBJECT_EXTENDED_LIMIT_INFORMATION, ctypes.byref(limits), ctypes.sizeof(limits), ): err = ctypes.get_last_error() kernel32.CloseHandle(self.job_handle) self.job_handle = None raise OSError(f"SetInformationJobObject failed. Windows Error Code: {err}") return self def __exit__(self, exc_type, exc_val, exc_tb): if self.job_handle: kernel32.CloseHandle(self.job_handle) self.job_handle = None def run_unsafe_script(self, script_path: str) -> subprocess.CompletedProcess: """Запускает скрипт с гарантированной привязкой к Job Object до исполнения кода.""" cmd = [sys.executable, script_path] proc = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, creationflags=CREATE_SUSPENDED | CREATE_NO_WINDOW, ) h_process = None h_thread = None try: # 1. Хэндл процесса h_process = kernel32.OpenProcess( PROCESS_SET_QUOTA | PROCESS_TERMINATE, False, proc.pid ) if not h_process: raise OSError( f"OpenProcess failed (PID {proc.pid}). " f"Windows Error Code: {ctypes.get_last_error()}" ) # 2. Хэндл главного потока tid = _find_main_thread_id(proc.pid) h_thread = kernel32.OpenThread(THREAD_SUSPEND_RESUME, False, tid) if not h_thread: raise OSError( f"OpenThread failed (TID {tid}). " f"Windows Error Code: {ctypes.get_last_error()}" ) # 3. Привязка к Job Object ДО исполнения кода if not kernel32.AssignProcessToJobObject(self.job_handle, h_process): err = ctypes.get_last_error() if err == ERROR_ACCESS_DENIED: raise OSError( "AssignProcessToJobObject вернул ERROR_ACCESS_DENIED. " "Возможные причины: процесс уже привязан к другому Job Object " "(вложенные Job Objects требуют Windows 8+ и флага " "JOB_OBJECT_LIMIT_SILENT_BREAKAWAY_OK); либо недостаточно прав " "на хэндл Job Object." ) raise OSError( f"AssignProcessToJobObject failed. Windows Error Code: {err}" ) # 4. Возобновление только после успешной привязки resumed = kernel32.ResumeThread(h_thread) if resumed == 0xFFFFFFFF: raise OSError( f"ResumeThread failed. " f"Windows Error Code: {ctypes.get_last_error()}" ) except Exception: proc.kill() raise finally: if h_thread: kernel32.CloseHandle(h_thread) if h_process: kernel32.CloseHandle(h_process) try: stdout, stderr = proc.communicate(timeout=self.time_limit_ms / 1000.0) return subprocess.CompletedProcess( args=cmd, returncode=proc.returncode, stdout=stdout, stderr=stderr, ) except subprocess.TimeoutExpired: if self.job_handle: kernel32.TerminateJobObject(self.job_handle, 1) proc.kill() try: stdout, stderr = proc.communicate(timeout=2.0) except subprocess.TimeoutExpired: stdout, stderr = "", "" raise subprocess.TimeoutExpired( cmd, self.time_limit_ms / 1000.0, stdout, stderr )
Python:import os import sys import time import subprocess import pytest pytestmark = pytest.mark.skipif( sys.platform != "win32", reason="Модуль windows_sandbox работает только на Windows.", ) from sandbox.windows_sandbox import WindowsSandbox def _write(tmp_path, name: str, body: str) -> str: p = tmp_path / name p.write_text(body, encoding="utf-8") return str(p) def _pid_alive(pid: int) -> bool: import ctypes from ctypes import wintypes k = ctypes.WinDLL("kernel32", use_last_error=True) k.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] k.OpenProcess.restype = wintypes.HANDLE k.CloseHandle.argtypes = [wintypes.HANDLE] k.CloseHandle.restype = wintypes.BOOL k.GetExitCodeProcess.argtypes = [wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD)] k.GetExitCodeProcess.restype = wintypes.BOOL PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 h = k.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid) if not h: return False code = wintypes.DWORD() alive = False if k.GetExitCodeProcess(h, ctypes.byref(code)): alive = (code.value == 259) k.CloseHandle(h) return alive def _extract_child_pid(text: str): for line in (text or "").splitlines(): if line.startswith("child_pid="): return int(line.split("=", 1)[1]) return None def test_basic_ok(tmp_path): script = _write(tmp_path, "ok.py", "print('hello')\n") with WindowsSandbox(time_limit_ms=5000) as sb: cp = sb.run_unsafe_script(script) assert cp.returncode == 0 assert cp.stdout.strip() == "hello" def test_basic_nonzero_exit(tmp_path): script = _write(tmp_path, "fail.py", "import sys; sys.exit(3)\n") with WindowsSandbox(time_limit_ms=5000) as sb: cp = sb.run_unsafe_script(script) assert cp.returncode == 3 def test_sleep_timeout(tmp_path): script = _write(tmp_path, "sleep.py", "import time; time.sleep(10)\n") t0 = time.monotonic() with WindowsSandbox(time_limit_ms=1000) as sb: with pytest.raises(subprocess.TimeoutExpired): sb.run_unsafe_script(script) elapsed = time.monotonic() - t0 assert elapsed < 5.0, f"Таймаут сработал слишком поздно: {elapsed:.2f}s" def test_child_tree_killed_on_timeout(tmp_path): body = ( "import subprocess, sys, time\n" "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n" "print(f'child_pid={child.pid}', flush=True)\n" "time.sleep(60)\n" ) script = _write(tmp_path, "tree.py", body) with WindowsSandbox(time_limit_ms=1500) as sb: with pytest.raises(subprocess.TimeoutExpired) as ei: sb.run_unsafe_script(script) child_pid = _extract_child_pid(ei.value.stdout) assert child_pid is not None, f"PID ребёнка не найден: {ei.value.stdout!r}" deadline = time.monotonic() + 3.0 while time.monotonic() < deadline and _pid_alive(child_pid): time.sleep(0.1) assert not _pid_alive(child_pid), ( f"Дочерний процесс {child_pid} пережил песочницу" ) def test_no_race_before_assign(tmp_path): marker = tmp_path / "RACE_MARKER" body = ( "import os\n" f"open(r'{marker}', 'w').write('x')\n" "print('marker_written', flush=True)\n" ) script = _write(tmp_path, "race.py", body) with WindowsSandbox(time_limit_ms=5000) as sb: cp = sb.run_unsafe_script(script) assert cp.returncode == 0 assert "marker_written" in cp.stdout assert marker.exists() def test_cpu_burn_killed_faster_than_wallclock(tmp_path): script = _write(tmp_path, "burn.py", "while True:\n pass\n") t0 = time.monotonic() with WindowsSandbox(time_limit_ms=500) as sb: with pytest.raises(subprocess.TimeoutExpired): sb.run_unsafe_script(script) elapsed = time.monotonic() - t0 assert elapsed < 5.0, f"CPU-лимит не сработал вовремя: {elapsed:.2f}s" def test_main_thread_marker(tmp_path): marker = tmp_path / "MAIN_MARKER" body = ( "import threading, time\n" f"open(r'{marker}', 'w').write('main')\n" "def bg():\n" " time.sleep(60)\n" "threading.Thread(target=bg).start()\n" "time.sleep(60)\n" ) script = _write(tmp_path, "main_thread.py", body) with WindowsSandbox(time_limit_ms=1000) as sb: with pytest.raises(subprocess.TimeoutExpired): sb.run_unsafe_script(script) assert marker.exists(), ( "Главный поток не выполнился — ResumeThread разбудил не тот TID" ) def test_no_deadlock_with_live_grandchild(tmp_path): body = ( "import subprocess, sys, time\n" "subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n" "time.sleep(60)\n" ) script = _write(tmp_path, "grandchild.py", body) t0 = time.monotonic() with WindowsSandbox(time_limit_ms=1000) as sb: with pytest.raises(subprocess.TimeoutExpired): sb.run_unsafe_script(script) elapsed = time.monotonic() - t0 assert elapsed < 5.0, f"Похоже на дедлок в communicate(): {elapsed:.2f}s" def test_invalid_time_limit_zero(): with pytest.raises(ValueError): WindowsSandbox(time_limit_ms=0) def test_invalid_time_limit_too_large(): with pytest.raises(ValueError): WindowsSandbox(time_limit_ms=86_400_001) def test_multiple_runs_in_one_sandbox(tmp_path): ok1 = _write(tmp_path, "ok1.py", "print('a')\n") ok2 = _write(tmp_path, "ok2.py", "print('b')\n") with WindowsSandbox(time_limit_ms=5000) as sb: cp1 = sb.run_unsafe_script(ok1) cp2 = sb.run_unsafe_script(ok2) assert cp1.stdout.strip() == "a" assert cp2.stdout.strip() == "b"
Практический кейс: Перечисление системных файлов через FindFirstFileW
Давайте протестируем систему боевой задачей. Передаем оркестратору ТЗ на низкоуровневый обход файлов в системной директории:
Что могло пойти не так при ручном написании?ЗАДАЧА: Написать скрипт на Python с использованием ctypes для перечисления всех файлов и директорий в системной папке Windows с помощью функций FindFirstFileW и FindNextFileW. Скрипт должен корректно определять типы объектов (файл или папка) и считать их общее количество.
ПРОВЕРКА_НАЛИЧИЯ: FindFirstFileW, FindNextFileW, FindClose, WIN32_FIND_DATAW, FILE_ATTRIBUTE_DIRECTORY
КОНТРАКТ_ВЫВОДА:
- Dir:.+
- File:.+
- Count:\s*\d+
ПОМНИ:
1. Путь поиска по умолчанию: "C:\\Windows\\System32\\*"
2. Структура WIN32_FIND_DATAW и FILETIME должны иметь DWORD-выравнивание (pack = 4), иначе ctypes упадет в access violation при записи.
3. Обязательно настроить argtypes и restype для FindFirstFileW.
4. Пропускать служебные директории "." и "..".
5. Хэндл поиска должен быть гарантированно закрыт через FindClose, чтобы избежать утечки ресурсов.
- Если забыть указать pack = 4 для структур WIN32_FIND_DATAW и FILETIME, ctypes выровняет поля по 8 байт. Как результат — FindFirstFileW запишет данные по неверным смещениям, и скрипт упадет в access violation writing.
- Если не выставить argtypes, путь передастся как ANSI-строка, хотя функция ожидает Wide-символы (LPCWSTR).
После запуска данного скрипта в изолированной песочнице, вывод полностью удовлетворил регулярным выражениям контракта (Dir:, File:, CountPython:import ctypes from ctypes import wintypes INVALID_HANDLE_VALUE = -1 FILE_ATTRIBUTE_DIRECTORY = 0x10 class FILETIME(ctypes.Structure): _pack_ = 4 _fields_ = [ ("dwLowDateTime", wintypes.DWORD), ("dwHighDateTime", wintypes.DWORD), ] class WIN32_FIND_DATAW(ctypes.Structure): _pack_ = 4 _fields_ = [ ("dwFileAttributes", wintypes.DWORD), ("ftCreationTime", FILETIME), ("ftLastAccessTime", FILETIME), ("ftLastWriteTime", FILETIME), ("nFileSizeHigh", wintypes.DWORD), ("nFileSizeLow", wintypes.DWORD), ("dwReserved0", wintypes.DWORD), ("dwReserved1", wintypes.DWORD), ("cFileName", wintypes.WCHAR * 260), ("cAlternateFileName", wintypes.WCHAR * 14), ] find_data = WIN32_FIND_DATAW() ctypes.windll.kernel32.FindFirstFileW.argtypes = [wintypes.LPCWSTR, ctypes.POINTER(WIN32_FIND_DATAW)] ctypes.windll.kernel32.FindFirstFileW.restype = wintypes.HANDLE ctypes.windll.kernel32.FindNextFileW.argtypes = [wintypes.HANDLE, ctypes.POINTER(WIN32_FIND_DATAW)] ctypes.windll.kernel32.FindNextFileW.restype = wintypes.BOOL ctypes.windll.kernel32.FindClose.argtypes = [wintypes.HANDLE] ctypes.windll.kernel32.FindClose.restype = wintypes.BOOL def get_target_data(): count_files = 0 count_dirs = 0 search_path = "C:\\Windows\\System32\\*" search_handle = ctypes.windll.kernel32.FindFirstFileW(search_path, ctypes.byref(find_data)) if search_handle == INVALID_HANDLE_VALUE: print("Error") return try: while True: name = find_data.cFileName if name not in (".", ".."): if find_data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY: print(f"Dir: {name}") count_dirs += 1 else: print(f"File: {name}") count_files += 1 if not ctypes.windll.kernel32.FindNextFileW(search_handle, ctypes.byref(find_data)): break finally: ctypes.windll.kernel32.FindClose(search_handle) print(f"Count: {count_files + count_dirs}") if __name__ == '__main__': get_target_data()
Финальный аудит безопасности от агента SecOps подтверждает успешное прохождение всех проверок:[
:]
Dir: .+File: .+- `Count: \s*\d+
### ВЕРДИКТ
APPROVED
Заключение
Использование связки из LLM-роя и низкоуровневых Job Objects позволяет решить две важнейшие задачи:
- Автоматизировать рутину написания капризного и чувствительного к типам данных ctypes кода под Windows.
- Гарантировать, что автоматический запуск сгенерированного кода не превратится в утечку ресурсов или деструктивный инцидент для хостовой операционной системы.
Итоги бета-теста локальных ИИ
Эксперимент доказывает: внедрение жестких шаблонов проектирования, нативной низкоуровневой подложки рантайм-валидации (WindowsSandbox) и слоев автоматического исправления синтаксиса позволяет добиться стабильной генерации системного софта от локальных моделей класса 7B–14B. Оркестратор успешно «докручивает» галлюцинации ИИ до полностью рабочего состояния, отсекая утечки ресурсов и падения в Access Violation.
Да, это узкоспециализированный конструктор, требующий калибровки под конкретные задачи и лимиты памяти. Тем не менее, как полигон для обкатки мультиагентных систем в автоматизации ИБ — это полностью прозрачный и рабочий стек.[/CODE][/SPOILER]
Код:
ЗАДАЧА: Написать скрипт на Python с использованием ctypes для перечисления всех файлов и директорий в системной папке Windows с помощью функций FindFirstFileW и FindNextFileW. Скрипт должен корректно определять типы объектов (файл или папка) и считать их общее количество.
ПРОВЕРКА_НАЛИЧИЯ: FindFirstFileW, FindNextFileW, FindClose, WIN32_FIND_DATAW, FILE_ATTRIBUTE_DIRECTORY
КОНТРАКТ_ВЫВОДА:
- Dir:.+
- File:.+
- Count:\s*\d+
ПОМНИ:
1. Путь поиска по умолчанию: "C:\\Windows\\System32\\*"
2. Структура WIN32_FIND_DATAW и FILETIME должны иметь DWORD-выравнивание (_pack_ = 4), иначе ctypes упадет в access violation при записи.
3. Обязательно настроить argtypes и restype для FindFirstFileW.
4. Пропускать служебные директории "." и "..".
5. Хэндл поиска должен быть гарантированно закрыт через FindClose, чтобы избежать утечки ресурсов.
Вложения
Последнее редактирование: