ОБСУЖДЕНИЕ

Автоматическая генерация и SecOps-валидация Windows ctypes-скриптов

0 ответов 42
Автор темы
Разбор мультиагентной системы Swarm Nexus: Автоматическая генерация и SecOps-валидация Windows ctypes-скриптов
Привет, Codeby!
Представляю архитектурный разбор конструктора-оркестратора на Python, разработанного для автоматического синтеза, декомпозиции и безопасного ревью низкоуровневого ИБ-инструментария (WinAPI/ctypes) с использованием локальных LLM (Qwen 2.5 Coder / Heretic).
swarm_nexus_2026.webp


Дисклеймер: Проект не является законченным коммерческим решением. Это гибкий экспериментальный конструктор-оркестратор, собиравшийся «под себя» (под конфигурацию и лимиты конкретного локального железа). Это наглядное пособие и бета-тест для понимания реальных границ возможностей локальных ИИ на данный момент развития технологий.
Система полностью автоматизирует цикл от анализа сырой идеи до генерации кода, динамического тестирования артефакта в изолированной подложке и прохождения финального SecOps-контроля.



Архитектурная схема пайплайна
Работа оркестратора разделена на 4 изолированных этапа, управляемых выделенными агентами через Ollama:
  1. Аналитик (analyst): Извлекает из описания задачи ключевые WinAPI-функции, формирует «критерии проверки» (AST-маркеры) и «контракт вывода» (Regex-паттерны для stdout).
  2. Кодер (coder): На основе ТЗ и строго типизированных безопасных шаблонов (Registry, Tokens, Process, File, Find) генерирует ctypes-код на Python.
  3. ОТК/Ревьюер (refactor): Итеративно отлавливает синтаксические ошибки, утечки дескрипторов (Handles), Access Violation и логические несоответствия контракту вывода. При зацикливании выполняет fallback на тяжелую рассуждающую модель (deepseek-r1:14b).
  4. Валидатор (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-код. Для решения этой проблемы развертывается конвейер из специализированных агентов, которые контролируют друг друга:
  1. Analyst (Аналитик): Принимает бизнес-задачу, проектирует строгое техническое задание, вычленяет необходимые для проверки WinAPI-функции и формирует жесткий контракт вывода (Output Contract) на базе регулярных выражений.
  2. Coder (Кодер): Специализированная модель, обученная правилам ctypes. Заполняет шаблон кода, настраивает сигнатуры функций, объявляет C-структуры Windows.
  3. Refactor (Рефактор): Подключается в случае падения скрипта на этапе runtime-тестирования. Анализирует Traceback и точечно устраняет WinAPI-специфичные баги.
  4. SecOps (Валидатор): Проводит статический аудит финального кода на полноту реализации контракта безопасности и отсутствие скрытых уязвимостей.
Ниже представлен исходный код управляющего оркестратора, который координирует работу агентов, применяет автоматические патчи (эвристики для ctypes-косяков вроде pack = 4) и гоняет код по итерациям исправления.

не влезает весь скрипт ниже в прикреплённых файлах переименуй в 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).
Оркестратор успешно прогнал генерацию, применил внутренние правила валидации и выдал следующий полностью рабочий и безопасный скрипт:

Python:
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()
После запуска данного скрипта в изолированной песочнице, вывод полностью удовлетворил регулярным выражениям контракта (Dir:, File:, Count:):

[
  • Dir: .+
  • File: .+
  • `Count: \s*\d+
:]
Финальный аудит безопасности от агента SecOps подтверждает успешное прохождение всех проверок:




Заключение
Использование связки из LLM-роя и низкоуровневых Job Objects позволяет решить две важнейшие задачи:
  1. Автоматизировать рутину написания капризного и чувствительного к типам данных ctypes кода под Windows.
  2. Гарантировать, что автоматический запуск сгенерированного кода не превратится в утечку ресурсов или деструктивный инцидент для хостовой операционной системы.
Делитесь в комментариях, какие проверки безопасности или специфичные WinAPI-патчи вы бы добавили в такой конвейер!





Итоги бета-теста локальных ИИ
Эксперимент доказывает: внедрение жестких шаблонов проектирования, нативной низкоуровневой подложки рантайм-валидации (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, чтобы избежать утечки ресурсов.
 

Вложения

Последнее редактирование: