import os, re, sys, logging, ast, tempfile, subprocess, time from typing import Dict, Any, Tuple, List import ollama if sys.platform == "win32": from sandbox.windows_sandbox import WindowsSandbox logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[logging.StreamHandler(sys.stdout)] ) MAX_ATTEMPTS = 3 MIN_CODE_LEN = 150 MIN_TZ_LEN = 50 RUNTIME_TIMEOUT = 15 OLLAMA_TIMEOUT = 180 KNOWN_WINAPI = [ "EnumWindows", "EnumChildWindows", "EnumThreadWindows", "EnumDesktopWindows", "IsWindowVisible", "IsWindow", "IsWindowEnabled", "GetWindowTextW", "GetWindowTextA", "GetWindowTextLengthW", "GetClassNameW", "GetClassNameA", "GetWindowThreadProcessId", "GetWindowRect", "GetClientRect", "GetForegroundWindow", "GetDesktopWindow", "FindWindowW", "FindWindowExW", "ShowWindow", "SetWindowPos", "MoveWindow", "CreateToolhelp32Snapshot", "Process32FirstW", "Process32NextW", "OpenProcess", "CloseHandle", "ReadProcessMemory", "WriteProcessMemory", "RegOpenKeyExW", "RegQueryValueExW", "RegCloseKey", "RegSetValueExW", "RegEnumKeyExW", "RegEnumValueW", "CreateFileW", "ReadFile", "WriteFile", "DeviceIoControl", "MapViewOfFile", "UnmapViewOfFile", "CreateFileMappingW", "VirtualAlloc", "VirtualFree", "VirtualProtect", "GetModuleHandleW", "GetProcAddress", "LoadLibraryW", "WINFUNCTYPE", "WNDENUMPROC", "HWND", "LPARAM", "WPARAM", "LRESULT", "POINT", "RECT", "MSG", "HANDLE", "DWORD", "BOOL", "OpenProcessToken", "LookupPrivilegeValueW", "GetTokenInformation", "GetCurrentProcess", "GetCurrentProcessId", "LookupPrivilegeNameW", "GetFileSizeEx", "GetFileAttributesW", "FindFirstFileW", "FindNextFileW", "FindClose", ] _WINAPI_NAMES_LOWER = set(k.lower() for k in KNOWN_WINAPI) _DLL_NAMES = { "kernel32", "user32", "advapi32", "ntdll", "gdi32", "shell32", "ole32", "comctl32", "comdlg32", "winmm", "ws2_32", "shcore", } CTYPES_FIXES = [ (r"wintypes\.ULONG_PTR", "ctypes.c_size_t"), (r"wintypes\.LONG_PTR", "ctypes.c_ssize_t"), (r"wintypes\.DWORD_PTR", "ctypes.c_size_t"), (r"wintypes\.SIZE_T", "ctypes.c_size_t"), (r"wintypes\.SSIZE_T", "ctypes.c_ssize_t"), (r"wintypes\.UINT_PTR", "ctypes.c_size_t"), (r"wintypes\.INT_PTR", "ctypes.c_ssize_t"), (r"\bwintypes\.c_size_t\b", "ctypes.c_size_t"), (r"\bwintypes\.c_ssize_t\b", "ctypes.c_ssize_t"), (r'\(\s*"HighPart"\s*,\s*wintypes\.LONG\s*\)\s*,\s*\(\s*"LowPart"\s*,\s*wintypes\.DWORD\s*\)', '("LowPart", wintypes.DWORD), ("HighPart", wintypes.LONG)'), (r"\(\s*'HighPart'\s*,\s*wintypes\.LONG\s*\)\s*,\s*\(\s*'LowPart'\s*,\s*wintypes\.DWORD\s*\)", "('LowPart', wintypes.DWORD), ('HighPart', wintypes.LONG)"), ] WINAPI_CONST_FIXES = [ (r"\bMISC_HKEY\s*=\s*0x2180\b", "MISC_HKEY = 0x80000001"), (r"\bHKEY_CURRENT_USER\s*=\s*0x2180\b", "HKEY_CURRENT_USER = 0x80000001"), (r"\bHKEY_LOCAL_MACHINE\s*=\s*0x2180\b", "HKEY_LOCAL_MACHINE = 0x80000002"), (r"\bHKEY_CURRENT_USER\s*=\s*0x8000001\b", "HKEY_CURRENT_USER = 0x80000001"), (r"(RegOpenKeyExW\s*\([^)]*?),\s*0x00020000\b", r"\1, 0x00020019"), (r'(key_path\s*=\s*r["\'][^"\']*)\\ShellState["\']', r'\1"'), (r'(RegOpenKeyExW\s*\([^,]+,\s*)b"([^"]*)"', r'\1"\2"'), (r'(RegQueryValueExW\s*\([^,]+,\s*)b"([^"]*)"', r'\1"\2"'), (r"INVALID_HANDLE_VALUE\s*=\s*-1\s*\((?:[^()]|\([^()]*\))*\)", "INVALID_HANDLE_VALUE = -1"), (r"ctypes\.windll\.toolhelp32\.", "ctypes.windll.kernel32."), (r"\.dwProcessId\b", ".th32ProcessID"), (r"\.cThreads\b", ".cntThreads"), (r"CreateToolhelp32Snapshot\s*\(\s*0x00020019\s*,", "CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS,"), (r"OpenProcessToken\s*\(\s*([^,]+),\s*TOKEN_ADJUST_PRIVILEGES\s*\|\s*TOKEN_QUERY\b", r"OpenProcessToken(\1, 0x0008"), (r"OpenProcessToken\s*\(\s*([^,]+),\s*TOKEN_ADJUST_PRIVILEGES\s*,", r"OpenProcessToken(\1, 0x0008,"), (r"OpenProcessToken\s*\(\s*([^,]+),\s*0x0020\s*,", r"OpenProcessToken(\1, 0x0008,"), (r"OpenProcessToken\s*\(\s*([^,]+),\s*(?:wintypes\.DWORD\s*\(\s*)?0x000A\s*\)?", r"OpenProcessToken(\1, 0x0008"), (r"OpenProcessToken\s*\(\s*([^,]+),\s*(?:wintypes\.DWORD\s*\(\s*)?10\s*\)?", r"OpenProcessToken(\1, 0x0008"), (r"OpenProcessToken\s*\(\s*([^,]+),\s*(?:wintypes\.DWORD\s*\(\s*)?0x0002\s*\)?", r"OpenProcessToken(\1, 0x0008"), (r"OpenProcessToken\s*\(\s*wintypes\.HANDLE\s*\(\s*-1\s*\)\s*,", "OpenProcessToken(ctypes.windll.kernel32.GetCurrentProcess(),"), (r"OpenProcessToken\s*\(\s*ctypes\.c_void_p\s*\(\s*-1\s*\)\s*,", "OpenProcessToken(ctypes.windll.kernel32.GetCurrentProcess(),"), (r"OpenProcessToken\s*\(\s*-\s*1\s*,", "OpenProcessToken(ctypes.windll.kernel32.GetCurrentProcess(),"), (r"CloseHandle\s*\(\s*ctypes\.windll\.kernel32\.GetCurrentProcess\(\)\s*\)", "# CloseHandle(GetCurrentProcess()) — не закрывать псевдо-хэндл"), (r"CloseHandle\s*\(\s*GetCurrentProcess\(\)\s*\)", "# CloseHandle(GetCurrentProcess()) — не закрывать псевдо-хэндл"), (r"(\(\"Privileges\"\s*,\s*LUID_AND_ATTRIBUTES\s*\*\s*)1(\))", r"\g<1>64\2"), (r"(\(\s*'Privileges'\s*,\s*LUID_AND_ATTRIBUTES\s*\*\s*)1(\))", r"\g<1>64\2"), (r"LookupPrivilegeNameW\s*\(\s*([^,]+),\s*([^,]+),\s*([^,]+),\s*None\s*\)", r"LookupPrivilegeNameW(\1, \2, \3, ctypes.byref(wintypes.DWORD(256)))"), (r"LookupPrivilegeValueW\s*\(\s*h_token\s*,", "LookupPrivilegeValueW(None,"), (r"LookupPrivilegeValueW\s*\(\s*hToken\s*,", "LookupPrivilegeValueW(None,"), (r"LookupPrivilegeValueW\s*\(\s*h_process\s*,", "LookupPrivilegeValueW(None,"), (r"^[ \t]*ctypes\.windll\.kernel32\.CloseHandle\.argtypes\s*=.*$", ""), (r"^[ \t]*ctypes\.windll\.kernel32\.CloseHandle\.restype\s*=.*$", ""), (r"^[ \t]*windll\.kernel32\.CloseHandle\.argtypes\s*=.*$", ""), (r"^[ \t]*windll\.kernel32\.CloseHandle\.restype\s*=.*$", ""), (r"^[ \t]*ctypes\.windll\.kernel32\.CloseHandle\.argtypes\s*=\s*\[[^\]]*\]\s*$", ""), (r'f"Privilege: SeDebugPrivilege\s*/\s*Status: (\w+)"', r'f"Privilege: SeDebugPrivilege\\nStatus: \1"'), (r'f"Privilege: (\w+)\s*/\s*Status: (\w+)"', r'f"Privilege: \1\\nStatus: \2"'), (r'"Privilege: SeDebugPrivilege / Status: (\w+)"', r'"Privilege: SeDebugPrivilege\nStatus: \1"'), (r'"Privilege: ([^"]+?)\s*/\s*Status: ([^"]+?)"', r'"Privilege: \1\\nStatus: \2"'), (r"'Privilege: ([^']+?)\s*/\s*Status: ([^']+?)'", r"'Privilege: \1\\nStatus: \2'"), ] _CONTRACT_PREFIX_RE = re.compile( r"^\s*(?:\*{0,2}|#{1,3}\s*)?(?:КОНТРАКТ[_ ]ВЫВОДА|Контракт\s+вывода)\s*:?\s*", re.IGNORECASE, ) _CRITERIA_PREFIX_RE = re.compile( r"^\s*(?:\*{0,2}|#{1,3}\s*)?(?:КРИТЕРИИ[_ ]ПРОВЕРКИ|Критерии\s+проверки)\s*:?\s*", re.IGNORECASE, ) _BAD_HEX_CONTRACT_RE = re.compile(r"^0x\[0-9a-fA-F\]\{\d+\}$") _LOOSE_HEX_CONTRACT_RE = re.compile(r"^\[0-9a-f \]\+$") _CANONICAL_HEX_CONTRACT = r"[0-9a-f]{2}( [0-9a-f]{2})+" _GLUED_SEP_RE = re.compile(r"(?:\\n|\s+0a\s+)") _OUTPUT_PREFIXES = ("PID:", "Name:", "Total:", "Length:", "HWND:", "Title:", "Class:", "SELF:", "Threads:", "Bytes:", "Value:", "Data:", "Privilege:", "Status:", "SeDebugPrivilege:", "Path:", "Size:", "Type:", "Count:", "Key:", "Subkey:", "File:", "Dir:") _HELP_MARKERS = ("или", "ctypes.sizeof", "wintypes.DWORD", "wintypes.HWND", "wintypes.BOOL", "wintypes.LPARAM", "wintypes.WCHAR", "ctypes.byref", "ctypes.create_", "=", "()") _PLACEHOLDER_MARKERS = ( "", "", "", "", "<Class>", "<Bytes>", "<Value>", "<Data>", "<Length>", "<Type>", "<Key>", "<Subkey>", "<Status>", "<Count>", "<SELF>", "<Threads>", "<Name>", "<File>", "<Path>", "<Size>", "—", "это", "через", "ОБЯЗАТЕЛЬНО", "НЕ используй", "используй", "второй аргумент", "третий аргумент", "первый аргумент", "где ", "например", "формат", "вывести", "выводит", "должен", "должна", "Помни", "ВАЖНО", "как ", "такой", "такая", "такое", "если", "флаг", "установлен", "не установлен", "в формате", "результат", "скрипт", ) _LITERAL_CONTRACT_PREFIXES = ( "Privilege:", "Status:", "Path:", "Size:", "PID:", "Name:", "Threads:", "SELF:", "Total:", "File:", "Count:", "Key:", "Subkey:", "HWND:", "Title:", "Class:", "Length:", "Bytes:", "Value:", "Data:", "Type:", "Dir:", ) _CYRILLIC_IN_PARENS_RE = re.compile(r"\(\s*[а-яА-ЯёЁ]") _CRITERIA_EXPRESSION_MARKERS = ( "ctypes.", "wintypes.", "windll.", "kernel32.", "user32.", "advapi32.", ) # Правка #10: маркеры процессных функций. Процессные паттерны (PID:, Name:, # Threads:, SELF:, Total:) добавляются в fallback ТОЛЬКО если в ТЗ есть # хотя бы одна из этих функций (или os.getpid для SELF). _PROCESS_FUNCTION_MARKERS = ( "createtoolhelp32snapshot", "process32firstw", "process32nextw", ) # Правка #11: пары «regex-паттерн + избыточный литерал». Если в контракте уже # есть regex слева, литерал справа удаляется (он покрывается regex'ом). _REDUNDANT_LITERAL_PAIRS = ( (r"Bytes: \d+", r"Bytes: 0"), (r"Status: \(Enabled\|Disabled\)", r"Status: Disabled"), (r"Status: \(Enabled\|Disabled\)", r"Status: Enabled"), (r"PID: \d+", r"PID: 0"), (r"Threads: \d+", r"Threads: 0"), ) _KNOWN_CONST_VALUES = { "token_query": ["0x0008", "0x8"], "token_privileges": ["3", "0x3"], "tokenprivileges": ["3", "0x3"], "se_privilege_enabled": ["0x2", "2"], "th32cs_snapprocess": ["0x2", "2"], "key_read": ["0x20019", "0x00020019"], "invalid_handle_value": ["-1"], "generic_read": ["0x80000000"], "file_share_read": ["0x1", "0x00000001"], "open_existing": ["3"], "file_attribute_directory": ["0x10", "16"], } _BAD_SECOPS_RECOMMENDATIONS = ( "ctypes.free_memory", "ctypes.free()", "c_char_p", "create_string_buffer должна быть заменена", "create_string_buffer заменить на c_char_p", "argtypes и restype", "argtypes/restype", "не настроена", "настроить argtypes", "добавить argtypes", "lookupprivilegevaluew не настроена", "c_size_t", "c_ssize_t", "size_t", "ssize_t", ) _HANDLE_LEAK_MARKERS = ( "# Removed CloseHandle", "Removed CloseHandle call", "# CloseHandle call removed", ) _MAX_REASONABLE_FILE_SIZE = 10**13 _TOKEN_USAGE_MARKERS = ( "OpenProcessToken", "GetTokenInformation", "LookupPrivilegeValueW", "LookupPrivilegeNameW", "AdjustTokenPrivileges", ) _TRUNCATED_RESPONSE_MARKERS = ( "unterminated string literal", "unexpected EOF", "EOF while scanning", "was never closed", ) # ============================================================ # ПРАВКА #20: нормализация пробелов в контракте + дедуп # ============================================================ _CONTRACT_PREFIXES_WITH_SPACE = ( "PID", "Name", "Threads", "Total", "SELF", "File", "Dir", "Count", "Key", "Subkey", "Bytes", "Path", "Size", "Privilege", "Status", "HWND", "Title", "Class", "Length", "Value", "Data", "Type", ) def _normalize_contract_spacing(patterns: List[str]) -> List[str]: """Правка #20: 'PID:\\d+' -> 'PID: \\d+', дедуп после нормализации.""" out = [] seen = set() for p in patterns: norm = p for pref in _CONTRACT_PREFIXES_WITH_SPACE: norm = re.sub(rf"^({re.escape(pref)})\s*:\s*", rf"\1: ", norm) if norm not in seen: seen.add(norm) out.append(norm) return out # ============================================================ # ПРАВКА #16: CP1251 -> UTF-8 мисдекод # ============================================================ _CYRILLIC_RE = re.compile(r"[а-яА-ЯёЁ]") _CJK_RE = re.compile(r"[\u4e00-\u9fff]") _MOJIBAKE_HINT_RE = re.compile(r"[\u0080-\u00ff\u4e00-\u9fff]{2,}") def _fix_cp1251_mojibake(text: str) -> str: """Пытается починить мисдекод cp1251-байт, прочитанных как utf-8/latin-1.""" if not text: return text if _CYRILLIC_RE.search(text) and not _CJK_RE.search(text): return text if not _MOJIBAKE_HINT_RE.search(text): return text candidates = [] try: candidates.append(text.encode("latin-1", errors="ignore").decode("cp1251", errors="ignore")) except Exception: pass try: candidates.append(text.encode("utf-8", errors="ignore").decode("cp1251", errors="ignore")) except Exception: pass best = text best_score = len(_CYRILLIC_RE.findall(text)) for cand in candidates: if not cand: continue score = len(_CYRILLIC_RE.findall(cand)) - len(_CJK_RE.findall(cand)) if score > best_score: best = cand best_score = score return best # ============================================================ # ПРАВКА #14: парсинг MISSING (без двоеточия, с ###-заголовком) # ============================================================ _MISSING_HEADER_RE = re.compile(r"^[#*\s]*MISSING\b[^\n]*$", re.IGNORECASE | re.MULTILINE) _MISSING_BULLET_RE = re.compile(r"^\s*[-*•]\s*(.+?)\s*$", re.MULTILINE) def _parse_missing_block(report: str) -> List[str]: if not report: return [] lines = report.splitlines() block_lines = [] in_block = False for ln in lines: if _MISSING_HEADER_RE.match(ln): in_block = True continue if in_block and re.match(r"^\s*###", ln): in_block = False continue if in_block: block_lines.append(ln) if block_lines: items = [] for ln in block_lines: m = _MISSING_BULLET_RE.match(ln) if m: items.append(m.group(1).strip().lower()) elif ln.strip() and len(ln.strip()) <= 80: items.append(ln.strip().lower()) if items: return items m = re.search(r"MISSING\s*:?\s*([^\n]+)", report, re.IGNORECASE) if m: raw = m.group(1) raw = re.split(r"[.)\]]", raw, maxsplit=1)[0] parts = [x.strip().lower() for x in re.split(r"[,;]", raw) if x.strip()] if parts: return parts return [] # ============================================================ # ПРАВКА #15: защита в глубину для __secops_fail_no_missing__ # ============================================================ def _is_missing_hallucination(report: str, code: str, missing: List[str]) -> bool: if not report or "missing" not in report.lower(): return False if missing: return False tail = re.split(r"missing", report, maxsplit=1, flags=re.IGNORECASE) tail = tail[1] if len(tail) > 1 else "" tokens = re.findall(r"[A-Za-z_][A-Za-z0-9_]{2,}", tail) if not tokens: return False code_low = (code or "").lower() return all(tok.lower() in code_low for tok in tokens) # ============================================================ # ПРАВКА #19: нормализация regex-паттернов контракта # ============================================================ _REGEX_NORMALIZE_PAIRS = [ (r"\+\+", "+"), (r"\*\*", "*"), (r"\?\?", "?"), ] def _normalize_contract_regex(pat: str) -> str: if not pat: return pat out = pat for bad, good in _REGEX_NORMALIZE_PAIRS: out = out.replace(bad, good) return out def _strip_contract_prefix(s: str) -> str: return _CONTRACT_PREFIX_RE.sub("", s).strip() def _strip_criteria_prefix(s: str) -> str: return _CRITERIA_PREFIX_RE.sub("", s).strip() def _sanitize_contract_pattern(p: str) -> str: if _BAD_HEX_CONTRACT_RE.match(p): logging.warning(f"[CONTRACT] Санитизация: '{p}' → '{_CANONICAL_HEX_CONTRACT}'") return _CANONICAL_HEX_CONTRACT if _LOOSE_HEX_CONTRACT_RE.match(p): logging.warning(f"[CONTRACT] Санитизация: '{p}' → '{_CANONICAL_HEX_CONTRACT}'") return _CANONICAL_HEX_CONTRACT return p def _split_glued_contract(p: str) -> List[str]: if not _GLUED_SEP_RE.search(p): return [p] parts = _GLUED_SEP_RE.split(p) result = [] for part in parts: part = part.strip().strip("'\"`«»„“”").strip() if part: result.append(part) if result: logging.info(f"[CONTRACT] Разбит склеенный паттерн на {len(result)} частей") return result def _is_real_contract(p: str) -> bool: if not p: return False if any(marker in p for marker in _PLACEHOLDER_MARKERS): return False if _CYRILLIC_IN_PARENS_RE.search(p): return False if any(p.startswith(pref) for pref in _OUTPUT_PREFIXES): return True if any(m in p for m in _HELP_MARKERS): return False if any(ch in p for ch in "\\[({+*?|^$"): return True return False def _is_literal_contract_candidate(p: str) -> bool: if not p: return False if any(marker in p for marker in _PLACEHOLDER_MARKERS): return False if _CYRILLIC_IN_PARENS_RE.search(p): return False if not any(p.startswith(pref) for pref in _LITERAL_CONTRACT_PREFIXES): return False return True def _group_alternative_patterns(patterns: List[str]) -> List[Any]: groups: Dict[str, List[str]] = {} order: List[str] = [] others: List[Any] = [] for p in patterns: m = re.match(r"^([A-Za-zА-Яа-я_]+:)\s+(.+)$", p) if m: prefix = m.group(1) value = m.group(2).strip() if value in ("Enabled", "Disabled", "True", "False", "Yes", "No", "OK", "FAIL", "PASS", "0", "1"): if prefix not in groups: groups[prefix] = [] order.append(prefix) groups[prefix].append(p) continue others.append(p) result: List[Any] = [] for prefix in order: if len(groups[prefix]) > 1: result.append(("ANY", groups[prefix])) logging.info(f"[CONTRACT] Альтернативы для '{prefix}': {groups[prefix]}") else: result.extend(groups[prefix]) result.extend(others) return result def _filter_criteria_by_task(criteria: List[str], task_keywords: List[str], tz_doc: str) -> List[str]: task_low = (tz_doc or "").lower() kw_low = set(k.lower() for k in task_keywords) result = [] for c in criteria: if c in kw_low: result.append(c) continue if re.search(r"\b" + re.escape(c) + r"\b", task_low): result.append(c) else: logging.warning(f"[CRITERIA] Отброшен критерий не из ТЗ: '{c}'") return result class Agent: def __init__(self, name: str, instruction: str, model: str): self.name = (name or "").strip() self.instruction = (instruction or "").strip() self.model = (model or "").strip() if not self.name or not self.model: raise ValueError("Agent: пустое имя или модель") class SwarmNexusOrchestrator: def __init__(self): self.analyst = Agent( "analyst", "", "R4C3R/qwen2.5-7b-instruct-heretic:q8_0" ) self.coder = Agent( "coder", "Ты Senior Python Developer, эксперт по ctypes и WinAPI.\n" "ТЫ ПИШЕШЬ ТОЛЬКО PYTHON. НЕ C, НЕ C++.\n" "ТВОЯ ЗАДАЧА: по ТЗ и шаблону вернуть ПОЛНЫЙ РАБОЧИЙ Python-код.\n" "ПРАВИЛА:\n" "1. ОБЯЗАН выдать блок ```python ... ``` (с закрывающими ```).\n" "2. WinAPI — через `ctypes.windll.user32.`, `ctypes.windll.kernel32.` или `ctypes.windll.advapi32.`.\n" "3. Строки из буферов — `ctypes.wstring_at(буфер)`.\n" "4. Коллбэки — `ctypes.WINFUNCTYPE`.\n" "5. Коллбэк `EnumWindows` — строго `ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HWND, wintypes.LPARAM)`.\n" "6. В `ctypes.wintypes` НЕТ `ULONG_PTR`/`LONG_PTR`/`DWORD_PTR`/`SIZE_T`/`c_size_t`/`c_ssize_t`. " "Используй `ctypes.c_size_t` / `ctypes.c_ssize_t`.\n" "7. РЕЕСТР:\n" " - `HKEY_CURRENT_USER = 0x80000001` (8 знаков).\n" " - `RegOpenKeyExW` — `lpSubKey` это путь к ключу, БЕЗ имени значения.\n" " - Имя значения — только в `RegQueryValueExW`, как `str`.\n" " - `samDesired = 0x00020019`, `RegQueryValueExW` — 6 аргументов.\n" " - `RegEnumKeyExW(hkey, index, name_buffer, byref(name_size), None, None, None, None)` — 8 аргументов.\n" " - `hkey = wintypes.HKEY()`, `RegOpenKeyExW(..., byref(hkey))`, `RegCloseKey(hkey)`.\n" " - Если в `КОНТРАКТ_ВЫВОДА` есть `Key: <literal>` (например, `Key: ShellState`), " "выводи БУКВАЛЬНО `print(f\"Key: ShellState\")`, а НЕ `print(f\"Key: {key}\")` из dict.\n" "8. ПРОЦЕССЫ:\n" " - `TH32CS_SNAPPROCESS = 0x2`, `INVALID_HANDLE_VALUE = -1`.\n" " - `CreateToolhelp32Snapshot` в `kernel32.dll`.\n" " - `PROCESSENTRY32W` объяви сам. `pe32.dwSize = ctypes.sizeof(PROCESSENTRY32W)` перед `Process32FirstW`.\n" " - `CloseHandle(snapshot)` ОБЯЗАТЕЛЬНО.\n" " - `count += 1` для ВСЕХ процессов — ЗА пределами `if/else`.\n" "9. ТОКЕНЫ:\n" " - `TOKEN_QUERY = 0x0008`, `TokenPrivileges = 3`, `SE_PRIVILEGE_ENABLED = 0x00000002`.\n" " - `GetCurrentProcess.restype = wintypes.HANDLE` ПЕРЕД вызовом.\n" " - `OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, byref(h_token))`.\n" " - `LUID`, `LUID_AND_ATTRIBUTES`, `TOKEN_PRIVILEGES` объяви сам. `Privileges` — `LUID_AND_ATTRIBUTES * 64`.\n" " - `LookupPrivilegeValueW(None, \"SeDebugPrivilege\", byref(luid))` — ПЕРВЫЙ АРГУМЕНТ `None`.\n" " - `CloseHandle(h_token)` ОБЯЗАТЕЛЕН.\n" " - Вывод — ДВЕ ОТДЕЛЬНЫЕ СТРОКИ через `\\n`.\n" "10. ФАЙЛЫ:\n" " - `CreateFileW(path, GENERIC_READ, FILE_SHARE_READ, None, OPEN_EXISTING, 0, None)`.\n" " - `CloseHandle(h_file)` ОБЯЗАТЕЛЬНО.\n" " - Для `GetFileSizeEx` используй `ctypes.c_longlong()` напрямую ИЛИ `LARGE_INTEGER` с `QuadPart`.\n" "11. ПЕРЕЧИСЛЕНИЕ ФАЙЛОВ (FindFirstFileW/FindNextFileW) — КРИТИЧЕСКИ ВАЖНО:\n" " - `find_data = WIN32_FIND_DATAW()` — СТРОГО структура, НЕ `create_string_buffer`.\n" " - `search_handle = FindFirstFileW(path, ctypes.byref(find_data))` — `byref` от структуры.\n" " - `FindNextFileW(search_handle, ctypes.byref(find_data))` — возвращает 0, когда файлов больше нет.\n" " - `FindClose(search_handle)` ОБЯЗАТЕЛЬНО.\n" " - `cFileName` — уже `str`, используй `find_data.cFileName` напрямую, НЕ `.value`, НЕ `.raw`.\n" " - `WIN32_FIND_DATAW` и `FILETIME` ДОЛЖНЫ иметь `_pack_ = 4` (DWORD-выравнивание). " "Без этого ctypes выравнивает по 8 и `FindFirstFileW` пишет по неверным смещениям → `access violation writing`.\n" " - ОБЯЗАТЕЛЬНО установи `argtypes`/`restype`: " "`FindFirstFileW.argtypes = [wintypes.LPCWSTR, ctypes.POINTER(WIN32_FIND_DATAW)]`, " "`FindFirstFileW.restype = wintypes.HANDLE`. Без этого ctypes передаёт путь как ANSI, " "а `FindFirstFileW` ждёт wide → мусор и `access violation`.\n" "12. ЗАПОЛНЯЙ ТОЛЬКО ТЕЛО ФУНКЦИИ ИЗ ШАБЛОНА. Весь код ВНУТРИ `def get_target_data():`. " "Не пиши код после `def` с отступом.\n" "13. ВСТАВЛЯЙ ТОЛЬКО ТЕ КОНСТАНТЫ И СТРУКТУРЫ, КОТОРЫЕ НУЖНЫ ДЛЯ ТЕКУЩЕЙ ЗАДАЧИ.\n" "14. НЕ КОПИРУЙ строки из блока 'Помни:' в код. Это справка.\n" "15. Начинай СРАЗУ с ```python.\n" "16. ПУТИ WINDOWS: используй обычную строку с двойными бэкслешами, например " "\"C:\\\\Windows\\\\System32\\\\*\" или r\"C:\\\\Windows\\\\System32\\\\*\". " "НЕ пиши r\"C:\\\" — сырая строка НЕ может заканчиваться на одиночный \\.\n" "17. FindFirstFileW: `find_data = WIN32_FIND_DATAW()` создаётся ОДИН РАЗ ДО цикла, " "FindFirstFileW пишет в него, затем в цикле `FindNextFileW(search_handle, byref(find_data))` " "перезаписывает ТУ ЖЕ структуру. НЕ создавай find_data заново внутри while.\n" "18. ФИЛЬТР ФАЙЛОВ: `if find_data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY:` — " "это ПАПКА, выводи `print(f\"Dir: {find_data.cFileName}\")`. " "`else:` — это ФАЙЛ, выводи `print(f\"File: {find_data.cFileName}\")`. " "Оба типа выводи, не фильтруй только файлы.\n" "19. В КОНЦЕ выведи `print(f\"Count: {count}\")` — суммарное количество (файлы + папки).\n" "20. Пропускай `.` и `..` при перечислении (они всегда присутствуют).\n" "21. Если функция печатает внутри и НЕ возвращает значение — в `__main__` вызывай " "`get_target_data()` БЕЗ `print()`, иначе в stdout попадёт `None`.\n" "22. `CloseHandle(h_token)`/`CloseHandle(h_file)`/`FindClose(search_handle)` — " "НЕ удаляй их никогда, даже если `argtypes` ломал код. Удаляй только `argtypes`.\n" "23. ПРОВЕРИТЬ vs ВКЛЮЧИТЬ: если ТЗ говорит «ПРОВЕРИТЬ наличие привилегии» — " "используй `GetTokenInformation(h_token, TokenPrivileges, buffer, buffer_size, byref(ret_len))` " "и проверку `Attributes & SE_PRIVILEGE_ENABLED`. " "НЕ используй `AdjustTokenPrivileges` — это ВКЛЮЧЕНИЕ привилегии, а не проверка. " "`OpenProcessToken` — с `TOKEN_QUERY`, НЕ с `TOKEN_ADJUST_PRIVILEGES`.\n" "24. ПОРЯДОК ОБЪЯВЛЕНИЙ: сначала `class LUID`, `class LUID_AND_ATTRIBUTES`, " "`class TOKEN_PRIVILEGES`, и только ПОТОМ `*.argtypes = [..., POINTER(TOKEN_PRIVILEGES), ...]`. " "Иначе `NameError: name 'TOKEN_PRIVILEGES' is not defined`.\n" "25. `GetTokenInformation` для `TokenPrivileges`: " "`buffer = ctypes.create_string_buffer(ctypes.sizeof(TOKEN_PRIVILEGES) + " "ctypes.sizeof(LUID_AND_ATTRIBUTES) * 64)`, " "`GetTokenInformation(h_token, TokenPrivileges, buffer, ctypes.sizeof(buffer), byref(ret_len))`, " "`tp = ctypes.cast(buffer, ctypes.POINTER(TOKEN_PRIVILEGES)).contents`, " "потом итерируй `tp.Privileges[:tp.PrivilegeCount]`.\n" "26. ПРОЦЕССЫ — ВЫВОДИ ТРИ СТРОКИ ДЛЯ КАЖДОГО:\n" " `print(f\"PID: {pe32.th32ProcessID}\")`\n" " `print(f\"Name: {pe32.szExeFile}\")`\n" " `print(f\"Threads: {pe32.cntThreads}\")`\n" " Если `pe32.szExeFile` пустой или `== \"System Idle Process\"` — пропусти (continue).\n" "27. SELF: `self_pid = os.getpid()`, и ЕСЛИ `pe32.th32ProcessID == self_pid` — " "выведи `print(f\"SELF: {self_pid}\")` ПЕРЕД строками PID/Name/Threads этого процесса " "(или один раз перед циклом — но тогда SELF будет до всех процессов, что тоже ок). " "`import os` ОБЯЗАТЕЛЕН.\n" "28. В КОНЦЕ — `print(f\"Total: {count}\")` — суммарное количество процессов " "(включая SELF, но БЕЗ пустых и System Idle Process).\n" "29. ОДИН ФАЙЛ (CreateFileW + GetFileSizeEx): НЕ перебирай файлы через FindFirstFileW, " "если ТЗ просит ОДИН файл. Открой `CreateFileW(path, GENERIC_READ, FILE_SHARE_READ, None, " "OPEN_EXISTING, 0, None)`, проверь `h_file != INVALID_HANDLE_VALUE`, вызови " "`GetFileSizeEx(h_file, ctypes.byref(size))`, выведи `Path:` и `Size:`, закрой `CloseHandle(h_file)`.\n" "30. Если в `КОНТРАКТ_ВЫВОДА` есть `Key: <literal>` (например, `Key: ShellState`), " "выводи БУКВАЛЬНО `print(f\"Key: ShellState\")`, а НЕ `print(f\"Key: {key}\")` из dict.\n" "31. Если ТЗ просит реестр (RegOpenKeyExW/RegQueryValueExW/RegCloseKey) — выводи " "`Key: <literal>` и `Bytes: <N>` БЕЗ циклов по dict, БЕЗ `Name:`, БЕЗ `Total:`.\n" "32. Если ТЗ говорит «файлы И папки» — выводи ОБА типа: " "`if find_data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY: print(f\"Dir: ...\")` " "`else: print(f\"File: ...\")`. НЕ фильтруй только файлы через `not (... & FILE_ATTRIBUTE_DIRECTORY)`.", "R4C3R/qwen2.5-coder-7b-instruct-heretic:q8_0" ) self.refactor = Agent( "refactor", "Ты Senior Python Developer. Исправь ТОЛЬКО указанную ошибку в коде ctypes.\n" "ВАЖНО: если `SyntaxError: unterminated string literal` и в коде есть `r\"C:\\\"` " "или ЛЮБАЯ `r\"...\"`, заканчивающаяся на одиночный `\\` — это и есть причина. " "Сырая строка НЕ может кончаться на `\\`. Замени на `\"C:\\\\\"` (обычная строка) " "или `r\"C:\\\\\"` (сырая с двойным бэкслешем в конце).\n" "ВАЖНО: если `NameError: name 'TOKEN_PRIVILEGES' is not defined` (или `LUID`, " "`LUID_AND_ATTRIBUTES`) — значит `*.argtypes = [..., POINTER(TOKEN_PRIVILEGES), ...]` " "стоит ВЫШЕ объявления `class TOKEN_PRIVILEGES`. Перемести строку `*.argtypes` " "ПОСЛЕ `class TOKEN_PRIVILEGES`. Если класса вообще НЕТ — добавь `class TOKEN_PRIVILEGES` " "ПЕРЕД первым `*.argtypes` с `TOKEN_PRIVILEGES`.\n" "ВАЖНО: если ТЗ говорит «ПРОВЕРИТЬ привилегию», а код вызывает `AdjustTokenPrivileges` — " "это НЕВЕРНО. Нужно `GetTokenInformation(h_token, TokenPrivileges, buffer, size, byref(ret_len))` " "и проверка `Attributes & SE_PRIVILEGE_ENABLED`. `AdjustTokenPrivileges` УБРАТЬ. " "`OpenProcessToken` — с `TOKEN_QUERY` (0x0008), НЕ с `TOKEN_ADJUST_PRIVILEGES`.\n" "ВАЖНО: если в коде только `print(f\"Total:\")` и `print(f\"SELF:\")`, а нет " "`print(f\"PID:\")`, `print(f\"Name:\")`, `print(f\"Threads:\")` — добавь их в цикл " "`while True:` ПЕРЕД `Process32NextW`, для каждого процесса (кроме пустых и " "`System Idle Process`).\n" "ВАЖНО: если `SELF: <pid>` выводится ПОСЛЕ `Total:` — перемести `print(f\"SELF: ...\")` " "ПЕРЕД циклом (или внутри цикла при совпадении PID с `os.getpid()`).\n" "ВАЖНО: если `count` не увеличивается — добавь `count += 1` в цикл ПЕРЕД `Process32NextW`.\n" "ВАЖНО: если `result_list.append(...)` используется для подсчёта — замени на `count += 1`.\n" "ВАЖНО: если `access violation writing` на `FindFirstFileW` — проверь две вещи: " "(1) `WIN32_FIND_DATAW` и `FILETIME` должны иметь `_pack_ = 4`, иначе ctypes выравнивает " "по 8 и `FindFirstFileW` пишет по неверным смещениям; " "(2) должны быть установлены `FindFirstFileW.argtypes = [wintypes.LPCWSTR, " "ctypes.POINTER(WIN32_FIND_DATAW)]` и `FindFirstFileW.restype = wintypes.HANDLE`, " "иначе путь передаётся как ANSI, а функция ждёт wide.\n" "ВАЖНО: если `access violation writing` и в коде `find_data = ctypes.create_string_buffer(...)` " "или `find_data = (ctypes.c_char * N)()` — замени на `find_data = WIN32_FIND_DATAW()`.\n" "ВАЖНО: если `SyntaxError: unexpected EOF` — верни ПОЛНЫЙ код, не обрезай строки.\n" "ВАЖНО: если `SyntaxError: unexpected indent` и в конце файла есть блок с отступом " "без `def` — удали этот блок.\n" "ВАЖНО: если `AttributeError: 'c_char_Array_N' object has no attribute 'value'` " "или `'raw'` — для `WIN32_FIND_DATAW.cFileName` используй `find_data.cFileName` напрямую.\n" "ВАЖНО: если ошибка «Утечка хэндла FindFirstFileW/FindClose» — добавь `FindClose(search_handle)`.\n" "ВАЖНО: если ошибка «Утечка хэндла OpenProcessToken/CloseHandle» — добавь `CloseHandle(h_token)`.\n" "ВАЖНО: если ошибка «Утечка хэндла CreateFileW/CloseHandle» — добавь `CloseHandle(h_file)`.\n" "ВАЖНО: если ошибка «Утечка хэндла CreateToolhelp32Snapshot/CloseHandle» — добавь " "`CloseHandle(snapshot)` после цикла.\n" "ВАЖНО: если `AttributeError: function 'CloseHandle' not found` — удали ТОЛЬКО " "`ctypes.windll.kernel32.CloseHandle.argtypes = ...` и `CloseHandle.restype`. " "Сам вызов `CloseHandle(...)` ОСТАВЬ. НЕ удаляй его — это утечка хэндла.\n" "ВАЖНО: если `Failed to adjust token privileges` — это потому что в ТЗ «проверить», " "а код «включает». Убери `AdjustTokenPrivileges`, используй `GetTokenInformation`.\n" "ВАЖНО: если `Privilege: ... / Status: ...` склеено через ` / ` — замени на две строки через `\\n`.\n" "ВАЖНО: если в stdout попал `None` в конце — убери `print()` вокруг вызова функции.\n" "ВАЖНО: если `AttributeError: 'c_longlong' object has no attribute 'QuadPart'` " "или `'HighPart'`/`'LowPart'` — используй `ctypes.c_longlong` напрямую: " "`size = ctypes.c_longlong()`, `total_size = size.value`. " "НЕ обращайся к `.QuadPart`, `.HighPart`, `.LowPart` у `c_longlong`.\n" "ВАЖНО: если `Size: N` и N > 10^13 — `LARGE_INTEGER` с полями в неправильном порядке. " "Порядок: `LowPart` (DWORD) первым, `HighPart` (LONG) вторым.\n" "ВАЖНО: НЕ добавляй `ctypes.free_memory(buffer)` или `ctypes.free()` для `create_string_buffer`.\n" "ВАЖНО: НЕ заменяй `create_string_buffer` на `c_char_p`.\n" "ВАЖНО: НЕ добавляй `argtypes`/`restype` для `LookupPrivilegeValueW`.\n" "ВАЖНО: если `Failed to lookup privilege value` — замени первый аргумент на `None`.\n" "ВАЖНО: если `IndexError: invalid index` и `Privileges` объявлен как " "`LUID_AND_ATTRIBUTES * 1` — замени на `LUID_AND_ATTRIBUTES * 64`.\n" "ВАЖНО: если `OpenProcessToken` второй аргумент `0x000A`/`0x0002`/`10`/" "`TOKEN_ADJUST_PRIVILEGES` — замени на `TOKEN_QUERY` (или `0x0008`).\n" "ВАЖНО: если в коде `CloseHandle(GetCurrentProcess())` — убери закрытие.\n" "ВАЖНО: если `GetTokenInformation` использует `byref(TOKEN_PRIVILEGES())` — " "замени на буфер `create_string_buffer(sizeof(TOKEN_PRIVILEGES) + " "sizeof(LUID_AND_ATTRIBUTES) * 64)` и `ctypes.cast(buffer, POINTER(TOKEN_PRIVILEGES))`.\n" "ВАЖНО: если `вывод не соответствует контракту` и в stdout УЖЕ есть ожидаемые строки — " "НЕ МЕНЯЙ код.\n" "ВАЖНО: если паттерн содержит `0a`/`\\n`/`или`/`ctypes.sizeof`/`= -1 (` — это справка, " "верни БЕЗ ИЗМЕНЕНИЙ.\n" "ВАЖНО: если в коде `os.getpid()` без `import os` — добавь `import os`.\n" "ВАЖНО: если `if pid != os.getpid():` без `else: print(f\"SELF: {pid}\")` — добавь `else`.\n" "ВАЖНО: если `count` увеличивается только внутри `if pid != os.getpid():` — перенеси `count += 1` ЗА `if/else`.\n" "ВАЖНО: если `RegEnumKeyExW` / `RegEnumValueW` с менее 8 аргументами — нужно 8.\n" "ВАЖНО: если `FindFirstFileW` с путём БЕЗ `\\*` в конце или с голым `C:\\` — " "замени на `\"C:\\\\Windows\\\\System32\\\\*\"`.\n" "ВАЖНО: если `find_data = WIN32_FIND_DATAW()` создаётся ВНУТРИ while — вынеси его ДО цикла.\n" "ВАЖНО: если вывод не содержит `File:` или `Count:` — добавь `print(f\"File: {find_data.cFileName}\")` " "и `print(f\"Count: {count}\")`.\n" "ВАЖНО: если ТЗ просит ОДИН файл (CreateFileW + GetFileSizeEx), а код перебирает все файлы " "через FindFirstFileW — это НЕВЕРНО. Убери FindFirstFileW/FindNextFileW/FindClose, " "оставь ОДИН вызов CreateFileW для `\"C:\\\\Windows\\\\System32\\\\kernel32.dll\"`, " "GetFileSizeEx, выведи `Path:` и `Size:`, закрой `CloseHandle(h_file)`.\n" "ВАЖНО: если ТЗ просит реестр (RegOpenKeyExW/RegQueryValueExW/RegCloseKey), " "а код выводит `Key: {key}` из dict — замени на `print(f\"Key: ShellState\")` ЛИТЕРАЛОМ. " "Убери `Name:`, `Total:`, `SELF:` — они НЕ относятся к задаче реестра.\n" "ВАЖНО: если ТЗ просит «файлы И папки», а код фильтрует только файлы через " "`if not (find_data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY):` — замени на " "`if find_data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY: print(f\"Dir: ...\")` " "`else: print(f\"File: ...\")`. Оба типа выводи, `count += 1` для обоих.\n" "Выдай РОВНО ОДИН блок ```python ... ```. Без пояснений.", "R4C3R/qwen2.5-coder-7b-instruct-heretic:q8_0" ) self.secops = Agent("secops", "", "nexus_secops:latest") self.refactor_fallback_model = "deepseek-r1:14b" self.context_limit = 16384 # ---- Шаблоны ---- self.code_template_enum = ( "import ctypes\nfrom ctypes import wintypes\n\n" "def get_target_data():\n" " result_list = []\n" " CallbackType = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HWND, wintypes.LPARAM)\n" " def my_callback(hwnd, lParam):\n" " if not ctypes.windll.user32.IsWindowVisible(hwnd):\n" " return True\n" " buffer = ctypes.create_unicode_buffer(256)\n" " length = ctypes.windll.user32.GetWindowTextW(hwnd, buffer, 256)\n" " if length > 0:\n" " class_buffer = ctypes.create_unicode_buffer(256)\n" " got_class = ctypes.windll.user32.GetClassNameW(hwnd, class_buffer, 256)\n" " if got_class == 0:\n" " return True\n" " result_list.append({'hwnd': hwnd, 'title': buffer.value, 'class_name': class_buffer.value})\n" " return True\n" " ctypes.windll.user32.EnumWindows(CallbackType(my_callback), 0)\n" " return result_list\n\n" "if __name__ == '__main__':\n" " for item in get_target_data():\n" " print(f\"HWND: 0x{item['hwnd']:x}\")\n" " print(f\"Title: {item['title']}\")\n" " print(f\"Class: {item['class_name']}\")\n" "{total_line}" ) self.code_template_generic = ( "import ctypes\nfrom ctypes import wintypes\n\n" "def get_target_data():\n" " result_list = []\n" " raise NotImplementedError('Заполни тело функции по ТЗ')\n\n" "if __name__ == '__main__':\n" " get_target_data()\n" ) self.code_template_process = ( "import ctypes\nimport os\nfrom ctypes import wintypes\n\n" "TH32CS_SNAPPROCESS = 0x2\n" "INVALID_HANDLE_VALUE = -1\n\n" "class PROCESSENTRY32W(ctypes.Structure):\n" " _fields_ = [\n" " (\"dwSize\", wintypes.DWORD),\n" " (\"cntUsage\", wintypes.DWORD),\n" " (\"th32ProcessID\", wintypes.DWORD),\n" " (\"th32DefaultHeapID\", ctypes.c_size_t),\n" " (\"th32ModuleID\", wintypes.DWORD),\n" " (\"cntThreads\", wintypes.DWORD),\n" " (\"th32ParentProcessID\", wintypes.DWORD),\n" " (\"pcPriClassBase\", wintypes.LONG),\n" " (\"dwFlags\", wintypes.DWORD),\n" " (\"szExeFile\", wintypes.WCHAR * 260),\n" " ]\n\n" "def get_target_data():\n" " result_list = []\n" " count = 0\n" " self_pid = os.getpid()\n" " print(f\"SELF: {self_pid}\")\n" " snapshot = ctypes.windll.kernel32.CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0)\n" " if snapshot == INVALID_HANDLE_VALUE:\n" " raise Exception(\"CreateToolhelp32Snapshot failed\")\n" " pe32 = PROCESSENTRY32W()\n" " pe32.dwSize = ctypes.sizeof(PROCESSENTRY32W)\n" " try:\n" " if not ctypes.windll.kernel32.Process32FirstW(snapshot, ctypes.byref(pe32)):\n" " raise Exception(\"Process32FirstW failed\")\n" " while True:\n" " name = pe32.szExeFile\n" " if name and name != \"System Idle Process\":\n" " print(f\"PID: {pe32.th32ProcessID}\")\n" " print(f\"Name: {name}\")\n" " print(f\"Threads: {pe32.cntThreads}\")\n" " count += 1\n" " if not ctypes.windll.kernel32.Process32NextW(snapshot, ctypes.byref(pe32)):\n" " break\n" " finally:\n" " ctypes.windll.kernel32.CloseHandle(snapshot)\n" " print(f\"Total: {count}\")\n\n" "if __name__ == '__main__':\n" " get_target_data()\n" ) self.code_template_token = ( "import ctypes\nfrom ctypes import wintypes\n\n" "TOKEN_QUERY = 0x0008\n" "TokenPrivileges = 3\n" "SE_PRIVILEGE_ENABLED = 0x00000002\n\n" "ctypes.windll.kernel32.GetCurrentProcess.restype = wintypes.HANDLE\n\n" "class LUID(ctypes.Structure):\n" " _fields_ = [(\"LowPart\", wintypes.DWORD), (\"HighPart\", wintypes.LONG)]\n\n" "class LUID_AND_ATTRIBUTES(ctypes.Structure):\n" " _fields_ = [(\"Luid\", LUID), (\"Attributes\", wintypes.DWORD)]\n\n" "class TOKEN_PRIVILEGES(ctypes.Structure):\n" " _fields_ = [(\"PrivilegeCount\", wintypes.DWORD), (\"Privileges\", LUID_AND_ATTRIBUTES * 64)]\n\n" "ctypes.windll.advapi32.OpenProcessToken.argtypes = [wintypes.HANDLE, wintypes.DWORD, ctypes.POINTER(wintypes.HANDLE)]\n" "ctypes.windll.advapi32.OpenProcessToken.restype = wintypes.BOOL\n" "ctypes.windll.advapi32.LookupPrivilegeValueW.argtypes = [wintypes.LPCWSTR, wintypes.LPCWSTR, ctypes.POINTER(LUID)]\n" "ctypes.windll.advapi32.LookupPrivilegeValueW.restype = wintypes.BOOL\n" "ctypes.windll.advapi32.GetTokenInformation.argtypes = [wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD)]\n" "ctypes.windll.advapi32.GetTokenInformation.restype = wintypes.BOOL\n\n" "def get_target_data():\n" " raise NotImplementedError('Заполни тело функции по ТЗ')\n\n" "if __name__ == '__main__':\n" " get_target_data()\n" ) self.code_template_registry = ( "import ctypes\nfrom ctypes import wintypes\n\n" "def get_target_data():\n" " HKEY_CURRENT_USER = 0x80000001\n" " KEY_READ = 0x00020019\n" " value_name = 'ShellState'\n" " subkey = r'Software\\\\Microsoft\\\\Windows\\\\CurrentVersion\\\\Explorer'\n" " hkey = wintypes.HKEY()\n" " if ctypes.windll.advapi32.RegOpenKeyExW(HKEY_CURRENT_USER, subkey, 0, KEY_READ, ctypes.byref(hkey)) != 0:\n" " print(f\"Key: ShellState\")\n" " print(f\"Bytes: 0\")\n" " return\n" " value_type = wintypes.DWORD()\n" " cbData = wintypes.DWORD(1024)\n" " buffer = ctypes.create_string_buffer(1024)\n" " ret = ctypes.windll.advapi32.RegQueryValueExW(hkey, value_name, None, ctypes.byref(value_type), buffer, ctypes.byref(cbData))\n" " if ret == 0:\n" " print(f\"Key: ShellState\")\n" " print(f\"Bytes: {cbData.value}\")\n" " else:\n" " print(f\"Key: ShellState\")\n" " print(f\"Bytes: 0\")\n" " ctypes.windll.advapi32.RegCloseKey(hkey)\n\n" "if __name__ == '__main__':\n" " get_target_data()\n" ) self.code_template_registry_enum = ( "import ctypes\nfrom ctypes import wintypes\n\n" "HKEY_CURRENT_USER = 0x80000001\n" "KEY_READ = 0x00020019\n" "ERROR_SUCCESS = 0\n" "ERROR_NO_MORE_ITEMS = 259\n\n" "def get_target_data():\n" " result_list = []\n" " subkey = r'Software\\\\Microsoft\\\\Windows\\\\CurrentVersion'\n" " hkey = wintypes.HKEY()\n" " if ctypes.windll.advapi32.RegOpenKeyExW(HKEY_CURRENT_USER, subkey, 0, KEY_READ, ctypes.byref(hkey)) != 0:\n" " raise Exception('Failed to open registry key')\n" " index = 0\n" " while True:\n" " name_buffer = ctypes.create_unicode_buffer(256)\n" " name_size = wintypes.DWORD(256)\n" " ret = ctypes.windll.advapi32.RegEnumKeyExW(hkey, index, name_buffer, ctypes.byref(name_size), None, None, None, None)\n" " if ret == ERROR_NO_MORE_ITEMS:\n" " break\n" " if ret != ERROR_SUCCESS:\n" " break\n" " result_list.append(name_buffer.value)\n" " index += 1\n" " ctypes.windll.advapi32.RegCloseKey(hkey)\n" " for name in result_list:\n" " print(f\"Key: {name}\")\n" " print(f\"Count: {len(result_list)}\")\n\n" "if __name__ == '__main__':\n" " get_target_data()\n" ) self.code_template_file = ( "import ctypes\nfrom ctypes import wintypes\n\n" "GENERIC_READ = 0x80000000\n" "FILE_SHARE_READ = 0x00000001\n" "OPEN_EXISTING = 3\n" "INVALID_HANDLE_VALUE = -1\n\n" "class LARGE_INTEGER(ctypes.Structure):\n" " _fields_ = [(\"QuadPart\", ctypes.c_longlong)]\n\n" "ctypes.windll.kernel32.CreateFileW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, ctypes.c_void_p, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE]\n" "ctypes.windll.kernel32.CreateFileW.restype = wintypes.HANDLE\n" "ctypes.windll.kernel32.GetFileSizeEx.argtypes = [wintypes.HANDLE, ctypes.POINTER(LARGE_INTEGER)]\n" "ctypes.windll.kernel32.GetFileSizeEx.restype = wintypes.BOOL\n\n" "def get_target_data():\n" " file_path = \"C:\\\\Windows\\\\System32\\\\kernel32.dll\"\n" " h_file = ctypes.windll.kernel32.CreateFileW(\n" " file_path, GENERIC_READ, FILE_SHARE_READ, None, OPEN_EXISTING, 0, None\n" " )\n" " if h_file == INVALID_HANDLE_VALUE:\n" " print(\"Error\")\n" " return\n" " try:\n" " size = LARGE_INTEGER()\n" " if not ctypes.windll.kernel32.GetFileSizeEx(h_file, ctypes.byref(size)):\n" " print(\"Error\")\n" " return\n" " print(f\"Path: {file_path}\")\n" " print(f\"Size: {size.QuadPart}\")\n" " finally:\n" " ctypes.windll.kernel32.CloseHandle(h_file)\n\n" "if __name__ == '__main__':\n" " get_target_data()\n" ) # ПРАВКА #18: code_template_find — выводим и файлы, и папки self.code_template_find = ( "import ctypes\nfrom ctypes import wintypes\n\n" "INVALID_HANDLE_VALUE = -1\n" "FILE_ATTRIBUTE_DIRECTORY = 0x10\n\n" "class FILETIME(ctypes.Structure):\n" " _pack_ = 4\n" " _fields_ = [\n" " (\"dwLowDateTime\", wintypes.DWORD),\n" " (\"dwHighDateTime\", wintypes.DWORD),\n" " ]\n\n" "class WIN32_FIND_DATAW(ctypes.Structure):\n" " _pack_ = 4\n" " _fields_ = [\n" " (\"dwFileAttributes\", wintypes.DWORD),\n" " (\"ftCreationTime\", FILETIME),\n" " (\"ftLastAccessTime\", FILETIME),\n" " (\"ftLastWriteTime\", FILETIME),\n" " (\"nFileSizeHigh\", wintypes.DWORD),\n" " (\"nFileSizeLow\", wintypes.DWORD),\n" " (\"dwReserved0\", wintypes.DWORD),\n" " (\"dwReserved1\", wintypes.DWORD),\n" " (\"cFileName\", wintypes.WCHAR * 260),\n" " (\"cAlternateFileName\", wintypes.WCHAR * 14),\n" " ]\n\n" "ctypes.windll.kernel32.FindFirstFileW.argtypes = [wintypes.LPCWSTR, ctypes.POINTER(WIN32_FIND_DATAW)]\n" "ctypes.windll.kernel32.FindFirstFileW.restype = wintypes.HANDLE\n" "ctypes.windll.kernel32.FindNextFileW.argtypes = [wintypes.HANDLE, ctypes.POINTER(WIN32_FIND_DATAW)]\n" "ctypes.windll.kernel32.FindNextFileW.restype = wintypes.BOOL\n" "ctypes.windll.kernel32.FindClose.argtypes = [wintypes.HANDLE]\n" "ctypes.windll.kernel32.FindClose.restype = wintypes.BOOL\n\n" "def get_target_data():\n" " count_files = 0\n" " count_dirs = 0\n" " find_data = WIN32_FIND_DATAW()\n" " search_path = \"C:\\\\Windows\\\\System32\\\\*\"\n" " search_handle = ctypes.windll.kernel32.FindFirstFileW(search_path, ctypes.byref(find_data))\n" " if search_handle == INVALID_HANDLE_VALUE:\n" " print(\"Error\")\n" " return\n" " try:\n" " while True:\n" " name = find_data.cFileName\n" " if name not in (\".\", \"..\"):\n" " if find_data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY:\n" " print(f\"Dir: {name}\")\n" " count_dirs += 1\n" " else:\n" " print(f\"File: {name}\")\n" " count_files += 1\n" " if not ctypes.windll.kernel32.FindNextFileW(search_handle, ctypes.byref(find_data)):\n" " break\n" " finally:\n" " ctypes.windll.kernel32.FindClose(search_handle)\n" " print(f\"Count: {count_files + count_dirs}\")\n\n" "if __name__ == '__main__':\n" " get_target_data()\n" ) # ===================== СЛУЖЕБНЫЕ МЕТОДЫ ===================== def _call_agent(self, agent: Agent, user_prompt: str, temp: float = 0.2, num_predict: int = 8192, num_ctx: int = None, timeout: int = None) -> str: try: messages = [] if agent.instruction: messages.append({"role": "system", "content": agent.instruction}) messages.append({"role": "user", "content": user_prompt}) options = { "temperature": float(temp), "num_ctx": int(num_ctx if num_ctx is not None else self.context_limit), "num_predict": int(num_predict), } chat_kwargs = {"model": agent.model, "messages": messages, "options": options} if timeout is not None: chat_kwargs["timeout"] = float(timeout) logging.info(f"[MODEL] Вызов '{agent.name}' → '{agent.model}' " f"(temp={temp}, num_predict={num_predict}, num_ctx={options['num_ctx']})") try: response = ollama.chat(**chat_kwargs) except TypeError: if "timeout" in chat_kwargs: chat_kwargs.pop("timeout", None) response = ollama.chat(**chat_kwargs) else: raise content = (response.get("message", {}) or {}).get("content", "") content = (content or "").strip() if not content: raise RuntimeError(f"Агент [{agent.name}] вернул пустой ответ") logging.info(f"[MODEL] '{agent.name}' ответил, длина={len(content)}") return content except Exception as e: logging.error(f"Сбой агента [{agent.name}]: {str(e)}") raise def _is_truncated_response(self, content: str) -> bool: if not content: return True if len(content.strip()) < MIN_CODE_LEN: return True low = content.lower() if any(m in low for m in _TRUNCATED_RESPONSE_MARKERS): return True return False def _code_is_syntactically_valid(self, code: str) -> Tuple[bool, str]: if not code or not code.strip(): return False, "пустой код" try: ast.parse(code) return True, "" except SyntaxError as e: return False, f"SyntaxError: {e.msg} (строка {e.lineno})" def _call_refactor_with_retry(self, refactor_prompt: str, force_fallback: bool = False) -> str: def _try_model(agent: Agent) -> str: content = self._call_agent(agent, refactor_prompt, temp=0.1, num_predict=4096, num_ctx=16384, timeout=OLLAMA_TIMEOUT) if self._is_truncated_response(content): raise RuntimeError("truncated response") pure = self._extract_pure_code(content) if not pure: raise RuntimeError("empty extracted code") ok, err = self._code_is_syntactically_valid(pure) if not ok: raise RuntimeError(f"invalid syntax response: {err}") return content if force_fallback: logging.warning(f"[REFACTOR] Принудительный fallback: {self.refactor_fallback_model}") fallback_agent = Agent("refactor_fallback", self.refactor.instruction, self.refactor_fallback_model) try: return _try_model(fallback_agent) except Exception as e: logging.error(f"[REFACTOR] Fallback тоже невалиден: {e}") return "" try: return _try_model(self.refactor) except Exception as e: err_str = str(e).lower() if "truncated response" in err_str: logging.warning("[REFACTOR] Основная модель вернула обрезанный ответ. Fallback.") elif "invalid syntax response" in err_str: logging.warning("[REFACTOR] Основная модель вернула код с SyntaxError. Fallback.") elif "пустой ответ" in err_str or "empty response" in err_str: logging.warning("[REFACTOR] Основная модель вернула пустой ответ. Fallback.") elif "timeout" in err_str or "timed out" in err_str: logging.warning("[REFACTOR] Основная модель превысила таймаут. Fallback.") elif "500" in err_str or "internal server error" in err_str: logging.warning("[REFACTOR] Ollama не загрузила модель. Жду 15 секунд.") time.sleep(15) try: return _try_model(self.refactor) except Exception: logging.warning("[REFACTOR] Повтор не удался. Fallback.") else: raise logging.warning(f"[REFACTOR] Использую fallback: {self.refactor_fallback_model}") fallback_agent = Agent("refactor_fallback", self.refactor.instruction, self.refactor_fallback_model) try: return _try_model(fallback_agent) except Exception as e: logging.error(f"[REFACTOR] Fallback тоже невалиден: {e}") return "" def _parse_keywords(self, raw: str) -> List[str]: if not raw: return [] kws = [] for kw in raw.split(","): kw = kw.strip().strip("'\"`«»„“”").strip().lower() if kw: kws.append(kw) return kws def _extract_keywords_from_task(self, task_text: str) -> List[str]: if not task_text: return [] matches = re.findall(r"ПРОВЕРКА_НАЛИЧИЯ:\s*(.+)", task_text, re.IGNORECASE) if matches: kws = self._parse_keywords(matches[-1]) logging.info(f"[KEYWORDS] Из task.txt: {kws}") return kws found = [] for kw in KNOWN_WINAPI: if re.search(r"\b" + re.escape(kw) + r"\b", task_text): found.append(kw.lower()) seen = set() result = [] for kw in found: if kw not in seen: seen.add(kw) result.append(kw) logging.info(f"[KEYWORDS] Авто: {result}") return result def _normalize_criterion(self, raw: str) -> List[str]: raw = raw.strip().lower() if not raw: return [] raw = _strip_criteria_prefix(raw) raw = raw.strip("'\"`«»„“”").strip() if not raw: return [] if raw in _DLL_NAMES: return [] if any(seg in raw for seg in _CRITERIA_EXPRESSION_MARKERS): logging.info(f"[CRITERIA] Отброшен критерий-выражение: '{raw}'") return [] if "." in raw and not raw.startswith("os."): logging.info(f"[CRITERIA] Отброшен критерий с точкой: '{raw}'") return [] if "_0x" in raw: parts = raw.split("_0x", 1) result = [parts[0]] if len(parts) > 1: result.append("0x" + parts[1]) return [p for p in result if p and len(p) >= 4 and p not in _DLL_NAMES] if " " in raw or len(raw) < 4: return [] return [raw] def _extract_criteria(self, tz_doc: str) -> List[str]: if not tz_doc: return [] header_re = re.compile( r"(?:\*{0,2}|#{1,3}\s*)?(?:КРИТЕРИИ[_ ]ПРОВЕРКИ|Критерии\s+проверки)\s*:?\s*\*{0,2}\s*\n", re.IGNORECASE ) m = header_re.search(tz_doc) criteria = [] if m: for line in tz_doc[m.end():].splitlines(): line = line.strip() if not line: if criteria: break continue if re.match(r"^#{1,3}\s", line): break if re.match(r"^\*{0,2}[А-ЯA-Z][А-ЯA-Z _]{3,}\s*:", line) and not _strip_criteria_prefix(line): break line = re.sub(r"^[-*•]\s*", "", line).strip() line = line.strip("'\"`«»„“”").strip().lower() if line: for part in self._normalize_criterion(line): if part not in criteria: criteria.append(part) if not criteria: for line in tz_doc.splitlines(): line = line.strip() if not line.startswith("- "): continue for part in self._normalize_criterion(line[2:].strip().lower()): if part not in criteria: criteria.append(part) logging.info(f"[CRITERIA] Извлечено (до фильтра): {criteria}") return criteria def _extract_output_contract(self, tz_doc: str) -> List[Any]: """Правка #4-fix/#4-fix-2/#4-fix-3/#10/#11/#19/#20.""" if not tz_doc: return [] header_re = re.compile( r"(?:\*{0,2}|#{1,3}\s*)?(?:КОНТРАКТ[_ ]ВЫВОДА|Контракт\s+вывода)\s*:?\s*\*{0,2}\s*\n", re.IGNORECASE ) m = header_re.search(tz_doc) raw_patterns = [] if m: for line in tz_doc[m.end():].splitlines(): line = line.strip() if not line: if raw_patterns: break continue if re.match(r"^#{1,3}\s", line): break if re.match(r"^\*{0,2}[А-ЯA-Z][А-ЯA-Z _]{3,}\s*:", line) and not _strip_contract_prefix(line): break line = re.sub(r"^[-*•]\s*", "", line).strip() line = _strip_contract_prefix(line).strip("'\"`«»„“”").strip() if line: raw_patterns.append(line) if not m: for line in tz_doc.splitlines(): candidate = _strip_contract_prefix(line.strip()) if candidate != line.strip(): candidate = candidate.strip("'\"`«»„“”").strip() if candidate and candidate not in raw_patterns: raw_patterns.append(candidate) task_low = (tz_doc or "").lower() has_process_functions = any(marker in task_low for marker in _PROCESS_FUNCTION_MARKERS) has_getpid = "os.getpid" in task_low or "getpid" in task_low if has_process_functions: if "pid" in task_low and not any("PID:" in p for p in raw_patterns): raw_patterns.append(r"PID: \d+") if ("name" in task_low or "szexefile" in task_low) and not any("Name:" in p for p in raw_patterns): raw_patterns.append(r"Name: \S+") if ("threads" in task_low or "cntthreads" in task_low) and not any("Threads:" in p for p in raw_patterns): raw_patterns.append(r"Threads: \d+") if "total" in task_low and not any("Total:" in p for p in raw_patterns): raw_patterns.append(r"Total: \d+") if has_getpid and "self" in task_low and not any("SELF:" in p for p in raw_patterns): raw_patterns.append(r"SELF: \d+") for line in tz_doc.splitlines(): cand = line.strip().strip("'\"`«»„“”").strip() if not cand: continue cand = re.sub(r"^[-*•]\s*", "", cand).strip() if not _is_literal_contract_candidate(cand): continue try: re.compile(cand) except re.error: continue if cand not in raw_patterns: raw_patterns.append(cand) if raw_patterns: logging.info(f"[CONTRACT] Fallback/дополнение из regex-строк: {raw_patterns}") expanded = [] for p in raw_patterns: for part in _split_glued_contract(p): if part not in expanded: expanded.append(part) raw_patterns = expanded filtered_patterns = [] for p in raw_patterns: redundant = False for regex_pat, literal_pat in _REDUNDANT_LITERAL_PAIRS: if p == literal_pat and any(rp == regex_pat for rp in raw_patterns): redundant = True logging.info(f"[CONTRACT] Удалён избыточный литерал '{p}' " f"(покрыт '{regex_pat}')") break if not redundant: filtered_patterns.append(p) raw_patterns = filtered_patterns # ПРАВКА #20: нормализация пробелов + дедуп ДО компиляции regex raw_patterns = _normalize_contract_spacing(raw_patterns) patterns = [] for p in raw_patterns: p = _normalize_contract_regex(p) # ПРАВКА #19 p = _sanitize_contract_pattern(p) if not _is_real_contract(p): logging.warning(f"[CONTRACT] Отброшено (справка, не паттерн): {p!r}") continue try: re.compile(p) if p not in patterns: patterns.append(p) except re.error: logging.warning(f"[CONTRACT] Невалидный regex: {p!r}") patterns = _group_alternative_patterns(patterns) logging.info(f"[CONTRACT] Извлечено (с группами): {patterns}") return patterns def _choose_template(self, keywords: List[str]) -> str: kw = set(k.lower() for k in keywords) if any(k in kw for k in ("openprocesstoken", "gettokeninformation", "lookupprivilegevaluew")): logging.info("[TEMPLATE] Token") return self.code_template_token if any(k in kw for k in ("createtoolhelp32snapshot", "process32firstw", "process32nextw")): logging.info("[TEMPLATE] Process") return self.code_template_process if any("enumwindows" in k for k in keywords): logging.info("[TEMPLATE] EnumWindows") return self.code_template_enum if any(k in kw for k in ("findfirstfilew", "findnextfilew", "findclose")): logging.info("[TEMPLATE] Find") return self.code_template_find if any(k in kw for k in ("createfilew", "getfilesizeex")): logging.info("[TEMPLATE] File") return self.code_template_file if any(k in ("regenumkeyexw", "regenumvaluew") for k in kw): logging.info("[TEMPLATE] RegistryEnum") return self.code_template_registry_enum if any(k in ("regopenkeyexw", "regqueryvalueexw", "regclosekey") for k in keywords): logging.info("[TEMPLATE] Registry") return self.code_template_registry logging.info("[TEMPLATE] Generic") return self.code_template_generic # ===================== ФИКСЫ ===================== def _fix_known_ctypes_issues(self, code: str) -> str: if not code: return code fixed = code for pattern, replacement in CTYPES_FIXES: new_fixed = re.sub(pattern, replacement, fixed) if new_fixed != fixed: logging.info(f"[FIX] '{pattern}' → '{replacement}'") fixed = new_fixed return fixed def _fix_known_winapi_constants(self, code: str) -> str: if not code: return code fixed = code for pattern, replacement in WINAPI_CONST_FIXES: new_fixed = re.sub(pattern, replacement, fixed, flags=re.MULTILINE) if new_fixed != fixed: logging.info(f"[FIX-CONST] '{pattern}' → '{replacement}'") fixed = new_fixed return fixed def _fix_raw_string_trailing_backslash(self, code: str) -> str: if not code: return code pattern = re.compile(r'(?<![A-Za-z0-9_])r(["\'])(.*?)(?<!\\)\\\1', re.DOTALL) def repl(m): quote = m.group(1) body = m.group(2) fixed_body = body.replace("\\", "\\\\") return f'{quote}{fixed_body}\\\\{quote}' new_code, n = pattern.subn(repl, code) if n: logging.info(f"[FIX-RAWSTR] Исправлено сырых строк с '\\' в конце: {n}") return new_code def _fix_c_longlong_attributes(self, code: str) -> str: if not code: return code vars_ = re.findall(r"(\w+)\s*=\s*ctypes\.c_longlong\s*\(\s*\)", code) if not vars_: return code fixed = code for var in set(vars_): for attr in ("QuadPart", "HighPart", "LowPart"): fixed = re.sub(rf"\b{re.escape(var)}\.{attr}\b", f"{var}.value", fixed) if fixed != code: logging.info(f"[FIX-CLONGLONG] Исправлены обращения к .QuadPart/.HighPart/.LowPart: {sorted(set(vars_))}") return fixed def _fix_large_integer_quadpart(self, code: str) -> str: if not code or "LARGE_INTEGER" not in code: return code m = re.search(r"class\s+LARGE_INTEGER\s*\(.*?\)\s*:(.*?)(?=\nclass |\ndef |\Z)", code, re.DOTALL) if not m: return code body = m.group(1) if "QuadPart" in body: return code fixed = re.sub(r"\b(\w+)\.QuadPart\b", r"((\1.HighPart << 32) | \1.LowPart)", code) if fixed != code: logging.info("[FIX-LARGEINT] .QuadPart у LARGE_INTEGER без QuadPart → (HighPart << 32) | LowPart") return fixed def _fix_find_path(self, code: str) -> str: if not code or "FindFirstFileW" not in code: return code fixed = code new_fixed = re.sub( r'(FindFirstFileW\s*\(\s*)(r?"C:\\\\?"|r?\'C:\\\\?\')(\s*,)', r'\1"C:\\\\Windows\\\\System32\\\\*"\3', fixed, ) if new_fixed != fixed: logging.info("[FIX-PATH] FindFirstFileW: голый 'C:\\' → 'C:\\\\Windows\\\\System32\\\\*'") fixed = new_fixed def add_star(m): prefix = m.group(1) path = m.group(2) suffix = m.group(3) if not path.endswith("\\*") and not path.endswith("*"): if path.endswith("\\") or path.endswith("/"): path = path + "*" else: path = path + "\\\\*" return f"{prefix}{path}{suffix}" new_fixed = re.sub( r'(FindFirstFileW\s*\(\s*)(r?"[^"]*"|r?\'[^\']*\')(\s*,)', add_star, fixed, ) if new_fixed != fixed: logging.info("[FIX-PATH] FindFirstFileW: добавлен '\\*' в конец пути") fixed = new_fixed return fixed def _fix_find_loop_structure(self, code: str) -> str: if not code or "FindFirstFileW" not in code: return code fixed = code lines = fixed.splitlines() out = [] i = 0 moved_find_data = False while i < len(lines): ln = lines[i] if re.match(r"^\s*find_data\s*=\s*WIN32_FIND_DATAW\s*\(\s*\)\s*$", ln): above = "\n".join(lines[:i]) if "FindFirstFileW" in above: moved_find_data = True i += 1 continue out.append(ln) i += 1 fixed = "\n".join(out) if moved_find_data: if not re.search(r"find_data\s*=\s*WIN32_FIND_DATAW\s*\(\s*\)", fixed): ffw_match = re.search(r"^([ \t]*)(.*FindFirstFileW.*)$", fixed, re.MULTILINE) if ffw_match: indent = ffw_match.group(1) fixed = (fixed[:ffw_match.start()] + f"{indent}find_data = WIN32_FIND_DATAW()\n" + fixed[ffw_match.start():]) logging.info("[FIX-LOOP] find_data вынесен из while перед FindFirstFileW") has_file_print = bool(re.search(r'print\s*\(\s*f?["\']File:\s*', fixed)) has_count_print = bool(re.search(r'print\s*\(\s*f?["\']Count:\s*', fixed)) if not has_file_print or not has_count_print: lines = fixed.splitlines() if not has_file_print and "cFileName" in fixed: added = False for idx, ln in enumerate(lines): if re.search(r"count\s*\+=\s*1", ln): indent = re.match(r"^([ \t]*)", ln).group(1) lines.insert(idx, f'{indent}print(f"File: {{find_data.cFileName}}")') added = True break if not added: for idx, ln in enumerate(lines): if re.search(r"if\s+not\s+.*FindNextFileW", ln): indent = re.match(r"^([ \t]*)", ln).group(1) lines.insert(idx, f'{indent}print(f"File: {{find_data.cFileName}}")') added = True break if added: logging.info("[FIX-LOOP] Добавлен print(f'File: ...')") if not has_count_print and "cFileName" in fixed: added_count = False for idx, ln in enumerate(lines): if "FindClose" in ln and "print" not in ln: indent = re.match(r"^([ \t]*)", ln).group(1) lines.insert(idx, f'{indent}print(f"Count: {{count}}")') added_count = True break if not added_count: lines.append(' print(f"Count: {count}")') if added_count: logging.info("[FIX-LOOP] Добавлен print(f'Count: ...')") fixed = "\n".join(lines) return fixed def _fix_process_output(self, code: str) -> str: if not code: return code if "Process32FirstW" not in code and "Process32NextW" not in code: return code fixed = code changed = False if "os.getpid" in fixed and not re.search(r"^\s*import\s+os\b", fixed, re.MULTILINE): lines = fixed.splitlines() insert_at = 0 for i, ln in enumerate(lines): if re.match(r"^\s*(import|from)\s+", ln): insert_at = i + 1 elif ln.strip() and not ln.strip().startswith("#"): break lines.insert(insert_at, "import os") fixed = "\n".join(lines) changed = True logging.info("[FIX-PROC] Добавлен 'import os'") has_pid = bool(re.search(r'print\s*\(\s*f?["\']PID:\s*', fixed)) has_name = bool(re.search(r'print\s*\(\s*f?["\']Name:\s*', fixed)) has_threads = bool(re.search(r'print\s*\(\s*f?["\']Threads:\s*', fixed)) has_self = bool(re.search(r'print\s*\(\s*f?["\']SELF:\s*', fixed)) has_total = bool(re.search(r'print\s*\(\s*f?["\']Total:\s*', fixed)) lines = fixed.splitlines() while_idx = None next_idx = None for i, ln in enumerate(lines): if re.match(r"^\s*while\s+True\s*:", ln): for j in range(i + 1, min(i + 50, len(lines))): if "Process32NextW" in lines[j]: while_idx = i next_idx = j break if while_idx is not None: break if while_idx is None or next_idx is None: logging.warning("[FIX-PROC] Не найден while True + Process32NextW — " "блок PID/Name/Threads НЕ вставлен. Проверь структуру кода.") else: if not (has_pid and has_name and has_threads): body_indent = None for j in range(while_idx + 1, next_idx): ln = lines[j] if ln.strip(): m = re.match(r"^([ \t]+)", ln) if m: body_indent = m.group(1) break if body_indent is None: body_indent = " " block = [ f'{body_indent}name = pe32.szExeFile', f'{body_indent}if name and name != "System Idle Process":', f'{body_indent} print(f"PID: {{pe32.th32ProcessID}}")', f'{body_indent} print(f"Name: {{name}}")', f'{body_indent} print(f"Threads: {{pe32.cntThreads}}")', f'{body_indent} count += 1', ] has_count_incr = any(re.search(r"count\s*\+=\s*1", lines[j]) for j in range(while_idx + 1, next_idx)) if has_count_incr: block = block[:-1] lines = lines[:next_idx] + block + lines[next_idx:] fixed = "\n".join(lines) changed = True logging.info("[FIX-PROC] Добавлен блок PID/Name/Threads в цикл перед Process32NextW") if not has_self and "os.getpid" not in fixed: lines = fixed.splitlines() for i, ln in enumerate(lines): if re.match(r"^\s*while\s+True\s*:", ln): indent = re.match(r"^([ \t]*)", ln).group(1) lines.insert(i, f'{indent}self_pid = os.getpid()') lines.insert(i + 1, f'{indent}print(f"SELF: {{self_pid}}")') fixed = "\n".join(lines) changed = True logging.info("[FIX-PROC] Добавлен 'SELF:' перед циклом") break if not has_total: lines = fixed.splitlines() inserted = False for i, ln in enumerate(lines): if "CloseHandle(snapshot)" in ln and "print" not in ln: indent = re.match(r"^([ \t]*)", ln).group(1) lines.insert(i, f'{indent}print(f"Total: {{count}}")') fixed = "\n".join(lines) changed = True logging.info("[FIX-PROC] Добавлен 'Total:' перед CloseHandle(snapshot)") inserted = True break if not inserted: fixed = fixed.rstrip() + '\n print(f"Total: {count}")\n' changed = True logging.info("[FIX-PROC] Добавлен 'Total:' в конец функции") return fixed def _fix_print_none(self, code: str) -> str: if not code or "def get_target_data" not in code: return code m = re.search(r"def get_target_data\s*\([^)]*\)\s*:(.*?)(?=\ndef |\nclass |\Z)", code, re.DOTALL) if not m: return code body = m.group(1) if re.search(r"^\s*return\b", body, re.MULTILINE): return code fixed = re.sub(r"print\s*\(\s*get_target_data\s*\(\s*\)\s*\)", "get_target_data()", code) if fixed != code: logging.info("[FIX-PRINT] Убран print() вокруг get_target_data()") return fixed def _fix_find_pack_and_argtypes(self, code: str) -> str: if not code or "FindFirstFileW" not in code: return code fixed = code if "class WIN32_FIND_DATAW" in fixed: tail = fixed.split("class WIN32_FIND_DATAW", 1)[1] head = tail.split("class ", 1)[0] if "_pack_" not in head: fixed = re.sub( r"(class\s+WIN32_FIND_DATAW\s*\([^)]*\)\s*:\s*\n)", r"\1 _pack_ = 4\n", fixed, ) logging.info("[FIX-FIND-PACK] Добавлен _pack_ = 4 в WIN32_FIND_DATAW") if "class FILETIME" in fixed: tail = fixed.split("class FILETIME", 1)[1] head = tail.split("class ", 1)[0] if "_pack_" not in head: fixed = re.sub( r"(class\s+FILETIME\s*\([^)]*\)\s*:\s*\n)", r"\1 _pack_ = 4\n", fixed, ) logging.info("[FIX-FIND-PACK] Добавлен _pack_ = 4 в FILETIME") if "FindFirstFileW.argtypes" not in fixed: m = re.search( r"(class\s+WIN32_FIND_DATAW\s*\(.*?\)\s*:\s*\n(?:\s+.*\n)*?\s*\]\s*\n)", fixed, re.DOTALL, ) if m: insert_at = m.end() block = ( "ctypes.windll.kernel32.FindFirstFileW.argtypes = " "[wintypes.LPCWSTR, ctypes.POINTER(WIN32_FIND_DATAW)]\n" "ctypes.windll.kernel32.FindFirstFileW.restype = wintypes.HANDLE\n" "ctypes.windll.kernel32.FindNextFileW.argtypes = " "[wintypes.HANDLE, ctypes.POINTER(WIN32_FIND_DATAW)]\n" "ctypes.windll.kernel32.FindNextFileW.restype = wintypes.BOOL\n" "ctypes.windll.kernel32.FindClose.argtypes = [wintypes.HANDLE]\n" "ctypes.windll.kernel32.FindClose.restype = wintypes.BOOL\n\n" ) fixed = fixed[:insert_at] + block + fixed[insert_at:] logging.info("[FIX-FIND-ARGTYPES] Добавлены argtypes/restype для FindFirstFileW/FindNextFileW/FindClose") return fixed def _fix_argtypes_order(self, code: str) -> str: if not code: return code if "TOKEN_PRIVILEGES" not in code and "LUID_AND_ATTRIBUTES" not in code and "LUID" not in code: return code fixed = code class_defs = { "LUID": ( "class LUID(ctypes.Structure):\n" " _fields_ = [(\"LowPart\", wintypes.DWORD), (\"HighPart\", wintypes.LONG)]\n" ), "LUID_AND_ATTRIBUTES": ( "class LUID_AND_ATTRIBUTES(ctypes.Structure):\n" " _fields_ = [(\"Luid\", LUID), (\"Attributes\", wintypes.DWORD)]\n" ), "TOKEN_PRIVILEGES": ( "class TOKEN_PRIVILEGES(ctypes.Structure):\n" " _fields_ = [(\"PrivilegeCount\", wintypes.DWORD), (\"Privileges\", LUID_AND_ATTRIBUTES * 64)]\n" ), } order = ["LUID", "LUID_AND_ATTRIBUTES", "TOKEN_PRIVILEGES"] for cls_name in order: if not re.search(rf"\b{cls_name}\b", fixed): continue cls_decl = re.search(rf"^class\s+{cls_name}\s*\(", fixed, re.MULTILINE) first_use = None for m in re.finditer(rf"\b{cls_name}\b", fixed): line_start = fixed.rfind("\n", 0, m.start()) + 1 line = fixed[line_start:fixed.find("\n", m.start()) if fixed.find("\n", m.start()) != -1 else len(fixed)] if re.match(rf"^\s*class\s+{cls_name}\s*\(", line): continue first_use = m.start() break if first_use is None: continue if cls_decl and cls_decl.start() < first_use: continue insert_pos = fixed.rfind("\n", 0, first_use) + 1 if cls_decl: decl_start = cls_decl.start() decl_end = cls_decl.end() rest = fixed[decl_end:] lines = rest.splitlines(keepends=True) consumed = 0 for ln in lines: if not ln.strip(): consumed += len(ln) continue if ln[0] not in (" ", "\t"): break consumed += len(ln) old_block = fixed[decl_start:decl_end + consumed] fixed = fixed[:decl_start] + fixed[decl_end + consumed:] if decl_start < insert_pos: insert_pos -= len(old_block) fixed = fixed[:insert_pos] + old_block + fixed[insert_pos:] logging.info(f"[FIX-ARGTYPES-ORDER] Перемещён class {cls_name} ПЕРЕД первым использованием") else: block = class_defs[cls_name] + "\n" fixed = fixed[:insert_pos] + block + fixed[insert_pos:] logging.info(f"[FIX-ARGTYPES-ORDER] Вставлен class {cls_name} ПЕРЕД первым использованием") return fixed def _fix_adjust_token_privileges(self, code: str) -> str: if not code or "AdjustTokenPrivileges" not in code: return code fixed = code lines = fixed.splitlines() out_lines = [] for ln in lines: if "AdjustTokenPrivileges" in ln and ("argtypes" in ln or "restype" in ln): continue out_lines.append(ln) fixed = "\n".join(out_lines) lines = fixed.splitlines() out_lines = [] i = 0 removed_any = False while i < len(lines): ln = lines[i] if "AdjustTokenPrivileges" in ln and "argtypes" not in ln and "restype" not in ln: indent = len(ln) - len(ln.lstrip()) i += 1 while i < len(lines): nxt = lines[i] if not nxt.strip(): i += 1 continue nxt_indent = len(nxt) - len(nxt.lstrip()) if nxt_indent > indent: i += 1 continue break removed_any = True continue out_lines.append(ln) i += 1 if removed_any: fixed = "\n".join(out_lines) logging.info("[FIX-ADJUST-TOKEN] Удалён вызов AdjustTokenPrivileges " "(ТЗ требует проверку, а не включение)") if "GetTokenInformation" not in fixed: m = re.search( r"^([ \t]*)(print\s*\(\s*f?[\"']Privilege:|print\s*\(\s*f?[\"']Status:)", fixed, re.MULTILINE, ) if m: indent = m.group(1) stub = ( f"{indent}# AdjustTokenPrivileges удалён: ТЗ требует ПРОВЕРКУ привилегии.\n" f"{indent}# GetTokenInformation не реализован — безопасный фолбэк:\n" f"{indent}print(\"Privilege: SeDebugPrivilege\")\n" f"{indent}print(\"Status: Disabled\")\n" f"{indent}return\n" ) fixed = fixed[:m.start()] + stub + fixed[m.start():] logging.info("[FIX-ADJUST-TOKEN] Вставлена заглушка 'Status: Disabled' " "(GetTokenInformation отсутствует)") return fixed def _ensure_import_os(self, code: str) -> str: if not code: return code if re.search(r"\bos\.", code) and not re.search(r"^\s*import\s+os\b", code, re.MULTILINE): logging.info("[FIX-IMPORT] Добавлен 'import os'") lines = code.splitlines() insert_at = 0 for i, ln in enumerate(lines): if re.match(r"^\s*(import|from)\s+", ln): insert_at = i + 1 elif ln.strip() and not ln.strip().startswith("#"): break lines.insert(insert_at, "import os") return "\n".join(lines) return code def _ensure_getcurrentprocess_restype(self, code: str) -> str: if not code or "GetCurrentProcess" not in code: return code lines = code.splitlines() call_idx = None restype_idx = None for i, ln in enumerate(lines): if call_idx is None and re.search(r"GetCurrentProcess\s*\(\s*\)", ln) and ".restype" not in ln: call_idx = i if restype_idx is None and "GetCurrentProcess.restype" in ln: restype_idx = i if call_idx is None: return code if restype_idx is not None and restype_idx < call_idx: return code restype_line = "ctypes.windll.kernel32.GetCurrentProcess.restype = wintypes.HANDLE" if restype_idx is not None: lines.pop(restype_idx) if restype_idx < call_idx: call_idx -= 1 lines.insert(call_idx, restype_line) logging.info("[FIX-RESTYPE] 'GetCurrentProcess.restype' вставлен ПЕРЕД вызовом") return "\n".join(lines) def _ensure_close_handle_for_token(self, code: str) -> str: if not code or "OpenProcessToken" not in code: return code if re.search(r"CloseHandle\s*\(\s*h_token\s*\)", code): return code lines = code.splitlines() in_func = False func_indent = None last_body_idx = None for i, ln in enumerate(lines): m = re.match(r"^(\s*)def\s+(\w+)\s*\(", ln) if m and "get_target_data" in ln: in_func = True func_indent = len(m.group(1)) continue if in_func: if not ln.strip(): continue cur_indent = len(ln) - len(ln.lstrip()) if cur_indent <= func_indent and not ln.lstrip().startswith("#"): break last_body_idx = i if last_body_idx is None: return code insert_line = " " * (func_indent + 4) + "ctypes.windll.kernel32.CloseHandle(h_token)" lines.insert(last_body_idx + 1, insert_line) logging.info("[FIX-HANDLE] Добавлен 'CloseHandle(h_token)'") return "\n".join(lines) def _ensure_close_handle_for_file(self, code: str) -> str: if not code or "CreateFileW" not in code: return code if re.search(r"CloseHandle\s*\(\s*h_file\s*\)", code): return code lines = code.splitlines() in_func = False func_indent = None last_body_idx = None for i, ln in enumerate(lines): m = re.match(r"^(\s*)def\s+(\w+)\s*\(", ln) if m and "get_target_data" in ln: in_func = True func_indent = len(m.group(1)) continue if in_func: if not ln.strip(): continue cur_indent = len(ln) - len(ln.lstrip()) if cur_indent <= func_indent and not ln.lstrip().startswith("#"): break last_body_idx = i if last_body_idx is None: return code insert_line = " " * (func_indent + 4) + "ctypes.windll.kernel32.CloseHandle(h_file)" lines.insert(last_body_idx + 1, insert_line) logging.info("[FIX-HANDLE] Добавлен 'CloseHandle(h_file)'") return "\n".join(lines) def _ensure_find_close(self, code: str) -> str: if not code or "FindFirstFileW" not in code: return code if re.search(r"FindClose\s*\(", code): return code lines = code.splitlines() in_func = False func_indent = None last_body_idx = None for i, ln in enumerate(lines): m = re.match(r"^(\s*)def\s+(\w+)\s*\(", ln) if m and "get_target_data" in ln: in_func = True func_indent = len(m.group(1)) continue if in_func: if not ln.strip(): continue cur_indent = len(ln) - len(ln.lstrip()) if cur_indent <= func_indent and not ln.lstrip().startswith("#"): break last_body_idx = i if last_body_idx is None: return code insert_line = " " * (func_indent + 4) + "ctypes.windll.kernel32.FindClose(search_handle)" lines.insert(last_body_idx + 1, insert_line) logging.info("[FIX-HANDLE] Добавлен 'FindClose(search_handle)'") return "\n".join(lines) def _remove_unused_token_structures(self, code: str) -> str: if not code: return code if any(marker in code for marker in _TOKEN_USAGE_MARKERS): return code lines = code.splitlines() out = [] skip_class = False class_indent = None for ln in lines: m = re.match(r"^(\s*)class\s+(LUID|LUID_AND_ATTRIBUTES|TOKEN_PRIVILEGES)\s*\(", ln) if m: skip_class = True class_indent = len(m.group(1)) continue if skip_class: if not ln.strip(): continue cur_indent = len(ln) - len(ln.lstrip()) if cur_indent <= class_indent: skip_class = False else: continue if re.match(r"^\s*(TOKEN_QUERY|TokenPrivileges|SE_PRIVILEGE_ENABLED)\s*=", ln): continue out.append(ln) result = "\n".join(out) result = re.sub(r"\n{3,}", "\n\n", result) if result != code: logging.info("[FIX-CLEAN] Удалены неиспользуемые токен-константы и структуры") return result def _apply_all_fixes(self, code: str) -> str: code = self._fix_raw_string_trailing_backslash(code) code = self._fix_c_longlong_attributes(code) code = self._fix_large_integer_quadpart(code) code = self._fix_known_ctypes_issues(code) code = self._fix_known_winapi_constants(code) code = self._fix_adjust_token_privileges(code) code = self._fix_argtypes_order(code) code = self._fix_find_path(code) code = self._fix_find_loop_structure(code) code = self._fix_process_output(code) code = self._fix_find_pack_and_argtypes(code) code = self._fix_print_none(code) code = self._ensure_import_os(code) code = self._ensure_getcurrentprocess_restype(code) code = self._ensure_close_handle_for_token(code) code = self._ensure_close_handle_for_file(code) code = self._ensure_find_close(code) code = self._remove_unused_token_structures(code) return code # ===================== SECOPS ===================== def _clean_secops_report(self, report: str) -> str: # ПРАВКА #16: сначала чиним CP1251-мисдекод, потом убираем \ufffd if not report: return "" report = _fix_cp1251_mojibake(report) cleaned = re.sub(r"[\u4e00-\u9fff\u3040-\u30ff\uac00-\ud7af\ufffd]", "", report) return re.sub(r"\n{3,}", "\n\n", cleaned).strip() def _parse_verdict(self, report: str) -> Tuple[bool, List[str]]: # ПРАВКА #14: парсим MISSING без двоеточия и с ###-заголовком if not report: return False, [] upper = report.upper() if "VERDICT: FAIL" in upper or re.search(r"\bFAIL\b", upper): missing = _parse_missing_block(report) if not missing: logging.warning("[SECOPS] FAIL без MISSING — вероятно, галлюцинация SecOps.") missing = ["__secops_fail_no_missing__"] return False, missing if "VERDICT: APPROVED" in upper or "APPROVED" in upper: return True, [] return False, [] def _validate_secops_missing(self, missing: List[str], pure_code: str, report: str = "") -> List[str]: # ПРАВКА #15: защита в глубину для __secops_fail_no_missing__ if not missing: return [] code_low = pure_code.lower() report_low = (report or "").lower() real_missing = [] for m in missing: if m == "__secops_fail_no_missing__": if any(bad in report_low for bad in _BAD_SECOPS_RECOMMENDATIONS): logging.warning("[SECOPS] FAIL без MISSING, но в отчёте есть " "посторонние рекомендации. Игнорирую.") continue if "найденные проблемы" in report_low: after = report_low.split("найденные проблемы", 1)[1] if "-" in after or "*" in after: logging.warning("[SECOPS] FAIL без MISSING, но есть 'найденные проблемы'. " "Считаю галлюцинацией, игнорирую.") continue # ПРАВКА #15: если все слова после 'missing' есть в коде — галлюцинация if _is_missing_hallucination(report, pure_code, missing): logging.warning("[SECOPS] MISSING без списка, но все токены после " "'missing' есть в коде. Считаю галлюцинацией, игнорирую.") continue real_missing.append(m) continue if m in code_low: logging.warning(f"[SECOPS] Галлюцинация: '{m}' есть в коде. Игнорирую.") continue if m in _WINAPI_NAMES_LOWER and m in code_low: logging.warning(f"[SECOPS] Галлюцинация WinAPI: '{m}' есть в коде. Игнорирую.") continue if m in _KNOWN_CONST_VALUES: if any(val in code_low for val in _KNOWN_CONST_VALUES[m]): logging.warning(f"[SECOPS] '{m}' есть как числовое значение. Игнорирую.") continue if any(bad in (report or "") for bad in _BAD_SECOPS_RECOMMENDATIONS): if m in ("argtypes", "restype", "argtypes/restype"): logging.warning(f"[SECOPS] '{m}' — это не критерий, игнорирую.") continue real_missing.append(m) return real_missing # ===================== ИЗВЛЕЧЕНИЕ КОДА ===================== def _extract_pure_code(self, raw_output: str) -> str: if not raw_output: return "" logging.info(f"[EXTRACT] raw len={len(raw_output)}") cleaned = re.sub(r"<think\b[^>]*>.*?", "", raw_output, flags=re.DOTALL | re.IGNORECASE) cleaned = re.sub(r"<think\b[^>]*>.*$", "", cleaned, flags=re.DOTALL | re.IGNORECASE) cleaned = re.sub(r"[\ufeff\u200b\u200c\u200d\u200e\u200f\u2060\u180e]", "", cleaned) code_blocks = re.findall( r"```[ \t]*(?:python|py)?[ \t]*\r?\n(.*?)\r?\n?[ \t]*```", cleaned, flags=re.DOTALL | re.IGNORECASE, ) if code_blocks: for block in reversed(code_blocks): b = block.strip() if ("def " in b or "class " in b) and "import " in b: return b return code_blocks[-1].strip() cleaned2 = re.sub(r"^[`]+", "", cleaned.strip()) first_nl = cleaned2.find("\n") if first_nl != -1: first_line = cleaned2[:first_nl].strip().lower() if first_line in ("python", "py"): cleaned2 = cleaned2[first_nl+1:] cleaned2 = re.sub(r"[`]+$", "", cleaned2).strip() if any(m in cleaned2 for m in ("def ", "import ", "ctypes", "class ")): return cleaned2 return "" def _get_ast_metrics(self, pure_code: str) -> Dict[str, Any]: metrics = {"calls": 0, "file_open_calls": 0, "has_placeholder": False} if not pure_code: return metrics code_no_comments = re.sub(r"#.*", "", pure_code) code_no_comments = re.sub(r'""".*?"""', "", code_no_comments, flags=re.DOTALL) code_no_comments = re.sub(r"'''.*?'''", "", code_no_comments, flags=re.DOTALL) stripped = re.sub(r"except[^\n]*:\s*\n\s*pass", "", code_no_comments) if re.search(r"^\s*pass\s*$", stripped, re.MULTILINE): metrics["has_placeholder"] = True if re.search(r"^\s*\.\.\.\s*$", code_no_comments, re.MULTILINE): metrics["has_placeholder"] = True if "raise notimplementederror" in code_no_comments.lower(): metrics["has_placeholder"] = True try: tree = ast.parse(pure_code) except SyntaxError: return metrics for node in ast.walk(tree): if isinstance(node, ast.Call): metrics["calls"] += 1 func = node.func if isinstance(func, ast.Name) and func.id == "open": metrics["file_open_calls"] += 1 if isinstance(func, ast.Attribute) and func.attr == "open": metrics["file_open_calls"] += 1 return metrics def _runtime_isolated_substrate_test(self, pure_code: str, contract: List[Any] = None) -> Tuple[bool, str]: if not pure_code or len(pure_code.strip()) < 10: return False, "Отказ: пустой код." with tempfile.NamedTemporaryFile(suffix=".py", delete=False, mode="w", encoding="utf-8", newline="\n") as tmp: tmp.write(pure_code) tmp_path = tmp.name try: if sys.platform == "win32": with WindowsSandbox(time_limit_ms=RUNTIME_TIMEOUT * 1000) as sandbox: result = sandbox.run_unsafe_script(tmp_path) else: result = subprocess.run([sys.executable, tmp_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=RUNTIME_TIMEOUT) output = (result.stdout or "") + (result.stderr or "") if result.returncode != 0: lines = [ln for ln in output.strip().splitlines() if ln.strip()] file_line = None for ln in lines: m = re.search(r'File ".*?", line (\d+)', ln) if m: file_line = m.group(1) err_lines = [ln for ln in lines if any( marker in ln for marker in ("Error", "Exception", "Traceback"))] last_err = err_lines[-1] if err_lines else (lines[-1] if lines else f"returncode={result.returncode}") if file_line: last_err = f"{last_err} (строка {file_line})" return False, f"Runtime Error: {last_err}" bad_markers = ("Traceback", "SyntaxError", "NameError", "TypeError", "AttributeError", "ValueError", "Exception", "FileNotFoundError", "IndexError", "OSError") if any(m in output for m in bad_markers): lines = [ln for ln in output.strip().splitlines() if ln.strip()] file_line = None for ln in lines: m = re.search(r'File ".*?", line (\d+)', ln) if m: file_line = m.group(1) err_lines = [ln for ln in lines if any( marker in ln for marker in ("Error", "Exception", "Traceback"))] last_err = err_lines[-1] if err_lines else (lines[-1] if lines else "unknown") if file_line: last_err = f"{last_err} (строка {file_line})" return False, f"Runtime Error: {last_err}" stdout_clean = (result.stdout or "").strip() if not stdout_clean: return False, "Runtime Error: нет вывода." if stdout_clean in ("[]", "{}", "None", "()"): return False, f"Runtime Error: пустой результат {stdout_clean}." if contract: for item in contract: if isinstance(item, tuple) and item[0] == "ANY": alts = item[1] found = any(re.search(a, stdout_clean, re.MULTILINE) for a in alts) if not found: preview = stdout_clean[:300].replace("\n", "\\n") return False, (f"Runtime Error: не найден ни один паттерн из группы " f"{alts} в stdout. Первые 300 символов: {preview!r}.") else: pattern = item try: if not re.search(pattern, stdout_clean, re.MULTILINE): preview = stdout_clean[:300].replace("\n", "\\n") return False, (f"Runtime Error: не найден паттерн `{pattern}` в stdout. " f"Первые 300 символов: {preview!r}.") except re.error as e: logging.warning(f"[CONTRACT] Ошибка компиляции {pattern!r}: {e}") size_match = re.search(r"Size:\s*(\d+)", stdout_clean) if size_match: size_val = int(size_match.group(1)) if size_val > _MAX_REASONABLE_FILE_SIZE: return False, (f"Runtime Error: Size={size_val} слишком большой (>10 ТБ). " f"Вероятно, `LARGE_INTEGER` объявлен с полями в неправильном порядке.") if size_val == 0: return False, "Runtime Error: Size=0 — вероятно, файл не открылся." return True, "Успешно выполнено." except subprocess.TimeoutExpired: return False, "Runtime Error: timeout (Job Object убил процесс)." except OSError as e: return False, f"Сбой подложки: {str(e)}" finally: if os.path.exists(tmp_path): try: os.remove(tmp_path) except OSError: pass def _normalize_err_for_loop(self, err_msg: str) -> str: return re.sub(r"Первые 300 символов.*$", "", err_msg, flags=re.DOTALL).strip() def _has_handle_leak(self, code: str) -> bool: if not code or "OpenProcessToken" not in code: return False if any(marker in code for marker in _HANDLE_LEAK_MARKERS): return True if re.search(r"OpenProcessToken\s*\([^,]+,\s*[^,]+,\s*ctypes\.byref\s*\(\s*h_token\s*\)\s*\)", code): if not re.search(r"CloseHandle\s*\(\s*h_token\s*\)", code): return True return False def _has_file_handle_leak(self, code: str) -> bool: if not code or "CreateFileW" not in code: return False if re.search(r"CreateFileW\s*\(", code): if not re.search(r"CloseHandle\s*\(\s*h_file\s*\)", code): return True return False def _has_find_handle_leak(self, code: str) -> bool: if not code or "FindFirstFileW" not in code: return False if re.search(r"FindFirstFileW\s*\(", code): if not re.search(r"FindClose\s*\(", code): return True return False # ===================== ОСНОВНОЙ ЦИКЛ ===================== def execute_swarm(self, user_idea: str) -> Dict[str, Any]: logging.info("=== ЗАПУСК NEXUS_SWARM_V4 ===") task_keywords = self._extract_keywords_from_task(user_idea) analyst_prompt = ( f"Спроектируй ТЗ для задачи: {user_idea}\n\n" f"ВАЖНО: критерии должны относиться ТОЛЬКО к ЭТОЙ задаче.\n\n" f"В конце добавь ДВА блока, каждый паттерн — С НОВОЙ СТРОКИ:\n\n" f"КРИТЕРИИ_ПРОВЕРКИ:\n" f"- функция или константа в коде\n\n" f"КОНТРАКТ_ВЫВОДА:\n" f"- regex-паттерн для строки stdout\n\n" f"Критерии — только имена функций/констант БЕЗ ТОЧЕК.\n" f"Контракт — regex БЕЗ кавычек, БЕЗ префиксов, каждый паттерн на ОТДЕЛЬНОЙ строке.\n" f"Для альтернатив (Enabled/Disabled) — ОДИН паттерн с группой." ) tz_doc = self._call_agent(self.analyst, analyst_prompt, temp=0.2, timeout=OLLAMA_TIMEOUT) if not tz_doc or len(tz_doc) < MIN_TZ_LEN: logging.error("ТЗ пустое.") return {"status": "FAILED_BY_TZ", "code": "", "secops_report": "", "tz": tz_doc, "contract": "(нет)"} logging.info("[OK] ТЗ получено.") criteria = self._extract_criteria(tz_doc) criteria = _filter_criteria_by_task(criteria, task_keywords, tz_doc) if not criteria: logging.warning("[CRITERIA] После фильтрации пусто.") contract = self._extract_output_contract(tz_doc) if not contract or len(contract) < 3: contract_from_task = self._extract_output_contract(user_idea) if contract_from_task and len(contract_from_task) > len(contract): contract = contract_from_task logging.info("[CONTRACT] Fallback на task.txt: извлечено " f"{len(contract)} паттернов") code_template = self._choose_template(task_keywords) # ПРАВКА #13: Total: в code_template_enum — только если в контракте есть Total: if code_template == self.code_template_enum and "{total_line}" in code_template: has_total = any(re.search(r"^\s*Total\s*:", c, re.IGNORECASE) for c in (contract or [])) total_line = ' print(f"Total: {len(result_list)}")' if has_total else "" code_template = code_template.format(total_line=total_line) criteria_hint = "\n".join(f"- {c}" for c in criteria) if criteria else "(нет)" contract_hint = "\n".join( (f"- ANY: {item[1]}" if isinstance(item, tuple) else f"- `{item}`") for item in contract ) if contract else "(нет)" user_prompt_coder = ( f"ТЗ: {tz_doc}\n\n" f"ШАБЛОН (структуру НЕ меняй, но ЗАПОЛНИ ТЕЛО функции!):\n```python\n{code_template}```\n\n" f"ВАЖНО: шаблон содержит `raise NotImplementedError` — ЗАМЕНИ его на реальную реализацию.\n" f"ВАЖНО: ЗАПОЛНЯЙ ТОЛЬКО ТЕЛО ФУНКЦИИ ИЗ ШАБЛОНА. Весь код ВНУТРИ `def get_target_data():`.\n" f"ВАЖНО: если ТЗ говорит «ПРОВЕРИТЬ привилегию» — используй `GetTokenInformation` " f"и проверку `Attributes & SE_PRIVILEGE_ENABLED`. НЕ используй `AdjustTokenPrivileges`. " f"`OpenProcessToken` — с `TOKEN_QUERY`.\n" f"ВАЖНО: `class TOKEN_PRIVILEGES` объявляется ДО `*.argtypes`, иначе `NameError`.\n" f"ВАЖНО: для `GetTokenInformation` — `buffer = create_string_buffer(sizeof(TOKEN_PRIVILEGES) + " f"sizeof(LUID_AND_ATTRIBUTES) * 64)`, `tp = cast(buffer, POINTER(TOKEN_PRIVILEGES)).contents`, " f"итерируй `tp.Privileges[:tp.PrivilegeCount]`.\n" f"ВАЖНО: для ПРОЦЕССОВ выводи `PID:`, `Name:`, `Threads:` для каждого, " f"`SELF:` перед циклом (или в цикле при совпадении), `Total:` в конце.\n" f"ВАЖНО: путь для FindFirstFileW — \"C:\\\\Windows\\\\System32\\\\*\" (двойные бэкслеши). " f"НЕ пиши r\"C:\\\" — сырая строка не может кончаться на \\.\n" f"ВАЖНО: `find_data = WIN32_FIND_DATAW()` создаётся ОДИН РАЗ ДО цикла, " f"а не внутри while.\n" f"ВАЖНО: `WIN32_FIND_DATAW` и `FILETIME` ДОЛЖНЫ иметь `_pack_ = 4`.\n" f"ВАЖНО: ОБЯЗАТЕЛЬНО установи `FindFirstFileW.argtypes = " f"[wintypes.LPCWSTR, ctypes.POINTER(WIN32_FIND_DATAW)]` и `restype = wintypes.HANDLE`.\n" f"ВАЖНО: `FindClose(search_handle)` ОБЯЗАТЕЛЕН.\n" f"ВАЖНО: `cFileName` — уже `str`, используй `find_data.cFileName`.\n" f"ВАЖНО: если ТЗ просит «файлы И папки» — выводи ОБА типа: " f"`if find_data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY: print(f\"Dir: ...\")` " f"`else: print(f\"File: ...\")`. НЕ фильтруй только файлы.\n" f"ВАЖНО: в конце `print(f\"Count: {{count}}\")` — суммарно (файлы + папки).\n" f"ВАЖНО: `CloseHandle(h_token)`/`CloseHandle(h_file)`/`CloseHandle(snapshot)` ОБЯЗАТЕЛЬНЫ. " f"НЕ удаляй их.\n" f"ВАЖНО: `GetCurrentProcess.restype = wintypes.HANDLE` ПЕРЕД вызовом.\n" f"ВАЖНО: `LookupPrivilegeValueW(None, ...)`, ПЕРВЫЙ АРГУМЕНТ `None`.\n" f"ВАЖНО: НЕ вызывай `ctypes.free_memory(buffer)` для `create_string_buffer`.\n" f"ВАЖНО: `get_target_data()` НЕ должна возвращать значение, если печатает внутри. " f"В `__main__` вызывай БЕЗ `print()`.\n" f"ВАЖНО: если ТЗ просит ОДИН файл (CreateFileW + GetFileSizeEx) — НЕ перебирай файлы " f"через FindFirstFileW. Открой ОДИН файл, получи размер, выведи `Path:` и `Size:`, закрой.\n" f"ВАЖНО: если в `КОНТРАКТ_ВЫВОДА` есть `Key: <literal>` (например, `Key: ShellState`), " f"выводи БУКВАЛЬНО `print(f\"Key: ShellState\")`, а НЕ `print(f\"Key: {{key}}\")` из dict.\n" f"ВАЖНО: если ТЗ просит реестр (RegOpenKeyExW/RegQueryValueExW/RegCloseKey) — выводи " f"`Key: <literal>` и `Bytes: <N>` БЕЗ циклов по dict, БЕЗ `Name:`, БЕЗ `Total:`.\n" f"Выдай ПОЛНЫЙ блок ```python ... ```." ) gen_out = self._call_agent(self.coder, user_prompt_coder, temp=0.2, timeout=OLLAMA_TIMEOUT) pure_code = self._extract_pure_code(gen_out) pure_code = self._apply_all_fixes(pure_code) attempt = 0 status = "REJECTED_UNKNOWN" err_msg = "Причина не установлена." prev_len = None prev_err_norm = None prev_code_norm = None force_fallback_next = False while attempt < MAX_ATTEMPTS: attempt += 1 logging.info(f"--- [ОТК] Попытка №{attempt} ---") ok_syntax, syntax_err = self._code_is_syntactically_valid(pure_code) if not ok_syntax: status = "REJECTED_RUNTIME" err_msg = f"SyntaxError: {syntax_err}" elif len(pure_code) < MIN_CODE_LEN or ("def " not in pure_code and "class " not in pure_code): status = "REJECTED_SABOTAGE" err_msg = "Код слишком короткий или нет функции/класса." elif self._has_handle_leak(pure_code): status = "REJECTED_SABOTAGE" err_msg = ("Утечка хэндла: `OpenProcessToken` есть, а `CloseHandle(h_token)` отсутствует. " "Верни `CloseHandle(h_token)` — это обязательный вызов.") elif self._has_file_handle_leak(pure_code): status = "REJECTED_SABOTAGE" err_msg = ("Утечка хэндла: `CreateFileW` есть, а `CloseHandle(h_file)` отсутствует.") elif self._has_find_handle_leak(pure_code): status = "REJECTED_SABOTAGE" err_msg = ("Утечка хэндла: `FindFirstFileW` есть, а `FindClose` отсутствует.") else: code_low = pure_code.lower() missing = [kw for kw in criteria if kw not in code_low] if missing: status = "REJECTED_SABOTAGE" err_msg = (f"В коде нет критериев: {', '.join(missing)}. " f"Добавь их. НЕ УДАЛЯЙ существующие функции.") else: metrics = self._get_ast_metrics(pure_code) if metrics["has_placeholder"]: status = "REJECTED_SABOTAGE" err_msg = "Есть pass/.../NotImplementedError. Заполни тело функции." elif metrics["file_open_calls"] > 0: status = "REJECTED_AST" err_msg = "Запрещён open()." else: success, err_msg = self._runtime_isolated_substrate_test(pure_code, contract) status = "APPROVED" if success else "REJECTED_RUNTIME" logging.info(f"[ВЕРДИКТ]: {status} | {err_msg}") if status == "APPROVED": break if attempt >= MAX_ATTEMPTS: logging.error(f"Провал после {MAX_ATTEMPTS}: {status}") return {"status": status, "tz": tz_doc, "code": pure_code, "secops_report": f"Провал после {MAX_ATTEMPTS}. {err_msg}", "contract": contract_hint} loop_warning = "" err_norm = self._normalize_err_for_loop(err_msg) code_norm = re.sub(r"\s+", " ", pure_code).strip() if prev_len is not None and prev_len == len(pure_code) and prev_err_norm == err_norm: loop_warning = ("\n\nВНИМАНИЕ: предыдущий ответ был ИДЕНТИЧЕН по длине и ошибке. " "Ты НЕ ИСПРАВИЛ код. Верни РАДИКАЛЬНО ДРУГОЙ код.") logging.warning("[LOOP] Зацикливание. Принудительный fallback на следующей итерации.") force_fallback_next = True # ПРАВКА #17: refactor вернул идентичный код -> сразу fallback if prev_code_norm is not None and prev_code_norm == code_norm: logging.warning("[REFACTOR] Идентичный код — принудительный fallback на deepseek-r1:14b") force_fallback_next = True refactor_prompt = ( f"ТЗ: {tz_doc}\n\nКРИТЕРИИ:\n{criteria_hint}\n\n" f"КОНТРАКТ:\n{contract_hint}\n\n" f"ОШИБКА: {err_msg}\n\nКОД:\n```python\n{pure_code}\n```\n\n" f"ЗАДАЧА: исправь. НЕ УДАЛЯЙ существующие функции.\n" f"КРИТИЧЕСКИ ВАЖНО: если в коде только `print(f\"Total:\")` и `print(f\"SELF:\")`, " f"а нет `print(f\"PID:\")`, `print(f\"Name:\")`, `print(f\"Threads:\")` — добавь их " f"в цикл ПЕРЕД `Process32NextW`, для каждого процесса (кроме пустых и " f"`System Idle Process`). `SELF:` — ПЕРЕД циклом. `Total:` — в конце.\n" f"КРИТИЧЕСКИ ВАЖНО: если `NameError: name 'TOKEN_PRIVILEGES' is not defined` — " f"`*.argtypes = [..., POINTER(TOKEN_PRIVILEGES), ...]` стоит ВЫШЕ `class TOKEN_PRIVILEGES`. " f"Перемести строку `*.argtypes` ПОСЛЕ `class TOKEN_PRIVILEGES`. Если класса нет — добавь его.\n" f"КРИТИЧЕСКИ ВАЖНО: если ТЗ «проверить привилегию», а код вызывает " f"`AdjustTokenPrivileges` — это НЕВЕРНО. Убери `AdjustTokenPrivileges`, " f"используй `GetTokenInformation` + проверку `Attributes & SE_PRIVILEGE_ENABLED`. " f"`OpenProcessToken` — с `TOKEN_QUERY`.\n" f"КРИТИЧЕСКИ ВАЖНО: если `Failed to adjust token privileges` — это потому что " f"код включает, а не проверяет. Убери `AdjustTokenPrivileges`.\n" f"КРИТИЧЕСКИ ВАЖНО: если `unterminated string literal` и в коде есть `r\"C:\\\"` — " f"замени на `\"C:\\\\\"`.\n" f"КРИТИЧЕСКИ ВАЖНО: если `access violation writing` на `FindFirstFileW` — " f"проверь `_pack_ = 4` и `argtypes`.\n" f"КРИТИЧЕСКИ ВАЖНО: если `find_data = ctypes.create_string_buffer(...)` — " f"замени на `find_data = WIN32_FIND_DATAW()`.\n" f"КРИТИЧЕСКИ ВАЖНО: если `SyntaxError: unexpected EOF` — верни ПОЛНЫЙ код.\n" f"КРИТИЧЕСКИ ВАЖНО: если `SyntaxError: unexpected indent` — удали блок без `def`.\n" f"КРИТИЧЕСКИ ВАЖНО: если `IndexError: invalid index` и `Privileges` — " f"`LUID_AND_ATTRIBUTES * 1` → `* 64`.\n" f"КРИТИЧЕСКИ ВАЖНО: если `OpenProcessToken` второй аргумент " f"`TOKEN_ADJUST_PRIVILEGES`/`0x0020`/`0x000A`/`10` — замени на `TOKEN_QUERY`/`0x0008`.\n" f"КРИТИЧЕСКИ ВАЖНО: если `Failed to lookup privilege value` — первый аргумент `None`.\n" f"КРИТИЧЕСКИ ВАЖНО: если `CloseHandle(GetCurrentProcess())` — убери закрытие.\n" f"КРИТИЧЕСКИ ВАЖНО: если вывод не содержит `Privilege:`/`Status:` — добавь " f"`print(f\"Privilege: SeDebugPrivilege\\nStatus: {{status}}\")`.\n" f"КРИТИЧЕСКИ ВАЖНО: если ТЗ просит ОДИН файл (CreateFileW + GetFileSizeEx), " f"а код перебирает все файлы через FindFirstFileW — УБЕРИ FindFirstFileW/" f"FindNextFileW/FindClose, оставь ОДИН вызов CreateFileW для " f"`\"C:\\\\Windows\\\\System32\\\\kernel32.dll\"`, GetFileSizeEx, выведи `Path:` " f"и `Size:`, закрой `CloseHandle(h_file)`.\n" f"КРИТИЧЕСКИ ВАЖНО: если ТЗ просит реестр (RegOpenKeyExW/RegQueryValueExW/RegCloseKey), " f"а код выводит `Key: {{key}}` из dict — замени на `print(f\"Key: ShellState\")` ЛИТЕРАЛОМ. " f"Убери `Name:`, `Total:`, `SELF:` — они НЕ относятся к задаче реестра.\n" f"КРИТИЧЕСКИ ВАЖНО: если ТЗ просит «файлы И папки», а код фильтрует только файлы " f"через `if not (find_data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY):` — замени на " f"`if find_data.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY: print(f\"Dir: ...\")` " f"`else: print(f\"File: ...\")`. Оба типа выводи, `count += 1` для обоих.\n" f"Выдай ТОЛЬКО блок ```python ... ```." f"{loop_warning}" ) prev_len = len(pure_code) prev_err_norm = err_norm prev_code_norm = code_norm try: gen_out = self._call_refactor_with_retry( refactor_prompt, force_fallback=force_fallback_next) except Exception as e: logging.error(f"[REFACTOR] Провал: {e}") return {"status": "REJECTED_REFACTOR_UNAVAILABLE", "tz": tz_doc, "code": pure_code, "secops_report": f"Refactor недоступен: {e}", "contract": contract_hint} force_fallback_next = False new_code = self._extract_pure_code(gen_out) if not new_code: logging.warning("[REFACTOR] Пустой ответ — оставляю старый код.") else: ok, _ = self._code_is_syntactically_valid(new_code) if not ok: logging.warning("[REFACTOR] Ответ не парсится — оставляю старый код.") else: pure_code = self._apply_all_fixes(new_code) criteria_text = "\n".join(f"- {c}" for c in criteria) if criteria else "(нет)" qa_prompt = ( f"ТЗ:\n{tz_doc}\n\nКРИТЕРИИ:\n{criteria_text}\n\n" f"КОД:\n```python\n{pure_code}\n```\n\n" f"Проверь, что в коде есть ВСЕ критерии (подстрока в нижнем регистре).\n" f"Если все есть:\nVERDICT: APPROVED\n\n" f"Если нет:\nVERDICT: FAIL\nMISSING: <через запятую конкретные критерии>\n\n" f"ВАЖНО: если не можешь указать КОНКРЕТНЫЙ критерий — верни APPROVED. НЕ FAIL без MISSING.\n" f"ВАЖНО: числовое значение константы (`0x0008` вместо `TOKEN_QUERY`) — НЕ MISSING.\n" f"ВАЖНО: НЕ требуй `argtypes`/`restype` — это НЕ критерий.\n" f"ВАЖНО: НЕ давай рекомендаций про `ctypes.free_memory`, `ctypes.free()`, `c_char_p`.\n" f"ВАЖНО: НЕ требуй дополнительной обработки ошибок WinAPI.\n" f"Ничего кроме VERDICT и MISSING. Две строки." ) qa_report = self._clean_secops_report( self._call_agent(self.secops, qa_prompt, temp=0.1, timeout=OLLAMA_TIMEOUT)) approved, missing_qa = self._parse_verdict(qa_report) if not approved and missing_qa: real_missing = self._validate_secops_missing(missing_qa, pure_code, qa_report) if not real_missing: logging.warning("[SECOPS] Все MISSING — галлюцинации или числовые эквиваленты. " "Принимаю код как APPROVED.") approved = True else: logging.warning(f"SecOps FAIL. Реально отсутствуют: {real_missing}") missing_qa = real_missing if not approved: logging.warning(f"SecOps FAIL. MISSING: {missing_qa}") return {"status": "REJECTED_BY_SECOPS", "tz": tz_doc, "code": pure_code, "secops_report": qa_report, "contract": contract_hint} logging.info("[OK] SecOps APPROVED.") out_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "output") os.makedirs(out_dir, exist_ok=True) with open(os.path.join(out_dir, "tz.md"), "w", encoding="utf-8") as f: f.write(tz_doc) with open(os.path.join(out_dir, "solution.py"), "w", encoding="utf-8") as f: f.write(pure_code) with open(os.path.join(out_dir, "secops_report.md"), "w", encoding="utf-8") as f: f.write(qa_report) with open(os.path.join(out_dir, "output_contract.md"), "w", encoding="utf-8") as f: f.write(contract_hint if contract else "(нет)") return {"status": "SUCCESS", "tz": tz_doc, "code": pure_code, "secops_report": qa_report, "contract": contract_hint} if __name__ == "__main__": try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except Exception: pass task_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "task.txt") if not os.path.exists(task_file): logging.error("Файл задачи не найден!") sys.exit(1) with open(task_file, "r", encoding="utf-8") as f: task_idea = f.read().strip() logging.info(f'Считано ТЗ: "{task_idea}"\n') orchestrator = SwarmNexusOrchestrator() res = orchestrator.execute_swarm(task_idea) print("\n==================== ФИНАЛЬНЫЙ РЕЗУЛЬТАТ РОЯ ====================") print(f"СТАТУС: {res.get('status')}\n") print(f"--- КОД ---\n{res.get('code')}\n") print(f"--- SECOPS ---\n{res.get('secops_report')}") print(f"--- CONTRACT ---\n{res.get('contract', '(нет)')}") # ===== КУСОК 2 КОНЕЦ =====