Разработка плагинов: Python Automation & C ABI Transformers¶
SORA спроектирована как открытая расширяемая платформа. Вы можете дополнять функционал аудитора на двух различных уровнях абстракции: 1. Высокоуровневые плагины автоматизации (Python): Для обработки событий сессии, интеграции с внешними API, отправки отчетов и автоматизации сценариев. 2. Низкоуровневые плагины модификации трафика (C ABI): Сверхбыстрые модули на C/Rust, встраиваемые непосредственно в сетевой конвейер PacketEngine для изменения байт пакетов на лету в режиме реального времени.
🐍 Высокоуровневые плагины на Python¶
Python-плагины используют событийную шину PluginBus и автоматически регистрируются менеджером PluginManager. При инициализации плагину передаются все необходимые зависимости (база данных, конфигурация).
Ниже приведен пример создания плагина отправки уведомлений в Telegram при захвате EAPOL-хэндшейка:
import structlog
from typing import Any
from sora.plugins.base import Plugin
logger = structlog.get_logger(__name__)
class TelegramNotifyPlugin(Plugin):
"""Отправляет уведомления о важных событиях аудита в Telegram."""
@property
def name(self) -> str:
return "telegram_notify"
def on_load(self, config: dict[str, Any]) -> None:
"""Вызывается при загрузке плагина. Получает настройки из TOML-профиля."""
self._token = config.get("bot_token")
self._chat_id = config.get("chat_id")
self._enabled = bool(config.get("enabled", False))
if self._enabled:
logger.info("telegram_plugin_initialized", chat_id=self._chat_id)
def on_event(self, event_name: str, data: dict[str, Any]) -> None:
"""Обрабатывает события, публикуемые ядром в PluginBus."""
if not self._enabled:
return
if event_name == "handshake_captured":
bssid = data.get("bssid")
client = data.get("client")
pcapng_path = data.get("pcapng_path")
message = f"🔑 Захвачен WPA2 хэндшейк!\nBSSID: {bssid}\nClient: {client}\nДамп: {pcapng_path}"
self._send_telegram_message(message)
def _send_telegram_message(self, text: str) -> None:
# Логика отправки HTTP-запроса к Telegram Bot API
logger.info("telegram_notification_sent", text=text)
⚙️ Низкоуровневые плагины-трансформеры (C ABI)¶
Для модификации трафика на лету во время инжекции (например, для динамического изменения полезной нагрузки или фаззинга) Python-слой слишком медленный. SORA предоставляет специальный C ABI v1 интерфейс трансформеров.
Скомпилированный .so плагин загружается через ctypes в Python и регистрируется в конвейере ядра Rust.
#ifndef SORA_TRANSFORMER_H
#define SORA_TRANSFORMER_H
#include <stdint.h>
#include <stddef.h>
// Таблица виртуальных методов (VTable) плагина-трансформера
typedef struct {
uint32_t abi_version;
const char* name;
// Инициализация плагина
int (*init)(const uint8_t* config_bytes, size_t config_len);
// Основной метод трансформации пакета
// Возвращает новую длину пакета, либо 0 для отброса пакета
size_t (*transform)(uint8_t* packet_buf, size_t packet_len, size_t max_buf_len);
// Деструктор плагина
void (*destroy)(void);
} SoraTransformerVTable;
// Экспортируемая точка входа, вызываемая SORA
extern const SoraTransformerVTable* get_sora_transformer(void);
#endif // SORA_TRANSFORMER_H
use std::os::raw::{c_char, c_int};
// Пример реализации трансформера, который инвертирует биты полезной нагрузки фреймов данных
static TRANSFORMER_NAME: *const c_char = b"PayloadInverter\0".as_ptr() as *const c_char;
#[no_mangle]
pub extern "C" fn get_sora_transformer() -> *const SoraTransformerVTable {
&VTABLE
}
static VTABLE: SoraTransformerVTable = SoraTransformerVTable {
abi_version: 1,
name: TRANSFORMER_NAME,
init: Some(transformer_init),
transform: Some(transformer_transform),
destroy: Some(transformer_destroy),
};
unsafe extern "C" fn transformer_init(_config: *const u8, _len: usize) -> c_int {
0 // Успех
}
unsafe extern "C" fn transformer_transform(buf: *mut u8, len: usize, _max_len: usize) -> usize {
let slice = std::slice::from_raw_parts_mut(buf, len);
// Инвертируем полезную нагрузку (пропускаем заголовок 802.11 - около 24 байт)
if len > 24 {
for byte in &mut slice[24..] {
*byte = !*byte;
}
}
len
}
unsafe extern "C" fn transformer_destroy() {}
🛠️ Регистрация плагина в TOML-профиле¶
Все плагины включаются и настраиваются непосредственно в TOML-файлах профилей аудита. При запуске сессии SORA автоматически инициализирует указанные плагины с переданными им параметрами.