Промышленные протоколы часто кажутся чёрным ящиком. Большинство туториалов предлагают просто pip install абстрактной библиотеки и вызов готовой функции. Но что на самом деле происходит на уровне байтов? Я решил разобраться: отключил все сторонние библиотеки, открыл raw-сокет на Python и вручную реализовал Modbus TCP. Результат — не production-клиент, а полное понимание того, как общаются программируемые логические контроллеры (ПЛК).

Почему Modbus TCP?

На современном заводе, станции водоочистки или в диспетчерской энергосети Modbus — это соединительная ткань для:

ПЛК (Programmable Logic Controllers)
Человеко-машинных интерфейсов (HMI)
SCADA-систем
Датчиков умных зданий

Несмотря на возраст, простота и совместимость держат Modbus в современных системах управления. А его структурная простота делает его идеальным полигоном для изучения промышленных протоколов.

Топология симуляции

Чтобы не рисковать реальным оборудованием, я развернул песочницу:

+-----------------------+
|     Хост-атакующий    |
|  (Скрипты на Python)  |
+-----------+-----------+
            |
            | TCP порт 502
            | (сырой поток)
            |
+-----------v-----------+
|    OpenPLC Runtime    |
|    (Целевой сервер)   |
+-----------------------+

Цель: OpenPLC с включённым Modbus TCP сервером на порту 502.

Рабочая станция: Arch Linux с tcpdump и Wireshark для верификации кадров.

Вместо pymodbus или Scapy я собирал каждый пакет вручную через модули socket и struct.

Анатомия протокола на уровне проводов

Стандартный запрос Modbus на проводе выглядит как плоская строка шестнадцатеричных данных:

00 01 00 00 00 06 01 03 00 00 00 01

Разобрав по байтам, видим 7-байтовый заголовок MBAP (Modbus Application Protocol) и PDU (Protocol Data Unit):

00 01 — Transaction Identifier: синхронизация запрос/ответ.
00 00 — Protocol Identifier: всегда 0 для Modbus TCP.
00 06 — Length: число оставшихся байт.
01 — Unit Identifier: адрес ПЛК.
03 — Function Code: команда (FC03 — Read Holding Registers).
00 00 — Starting Address: начальный адрес регистра.
00 01 — Quantity of Registers: количество 16-битных слов.

Как только эта структура стала понятна, перевод в бинарные данные стал тривиальным.

Фаза 1: Пассивный сбор данных (FC03 и FC01)

Чтение Holding Registers (FC03)

Первый скрипт собирал цикл транзакций FC03 для чтения аналоговых значений из ПЛК.

import socket
import struct

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('192.168.1.100', 502))

# Формируем запрос FC03: чтение 1 регистра с адреса 0
transaction_id = 1
protocol_id = 0
length = 6
unit_id = 1
function_code = 3
start_addr = 0
quantity = 1

# Упаковываем в бинарный формат
request = struct.pack('>HHHBBHH',
    transaction_id, protocol_id, length,
    unit_id, function_code, start_addr, quantity)

sock.send(request)
response = sock.recv(1024)
print('Ответ (hex):', response.hex())
# Ожидаем: 00 01 00 00 00 05 01 03 02 00 00
sock.close()

Увидев чистые байты на сокете, я перестал воспринимать протокол как магию.

Чтение Discrete Coils (FC01)

Реализовал FC01 для проверки битовых флагов. Отслеживание упакованных битов в формате LSB показало, насколько лёгким и предсказуемым является опрос состояний.

Фаза 2: Активное изменение состояния (FC06 и FC16)

Чтение позволяет мониторить, запись — менять физическую реальность.

Запись одного регистра (FC06)

Отправил значение 1337 в регистр 0 через Function Code 6:

import socket
import struct

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('192.168.1.100', 502))

transaction_id = 2
protocol_id = 0
length = 6
unit_id = 1
function_code = 6  # Write Single Register
register_addr = 0
value = 1337

request = struct.pack('>HHHBBHH',
    transaction_id, protocol_id, length,
    unit_id, function_code, register_addr, value)

sock.send(request)
response = sock.recv(1024)
print('Ответ на запись:', response.hex())
sock.close()

Считывание подтвердило мгновенное изменение в рантайме. Это напоминание, как легко неаутентифицированная инъекция меняет параметры контроллера.

Массовая запись (FC16)

С помощью FC16 (Write Multiple Registers) отправил массив значений:

import socket
import struct

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('192.168.1.100', 502))

# Данные: [100, 200, 300, 400, 500]
values = [100, 200, 300, 400, 500]
quantity = len(values)
byte_count = quantity * 2  # 2 байта на регистр

transaction_id = 3
protocol_id = 0
length = 9 + quantity * 2  # 7 MBAP + 2 (адрес+количество) + 1 (byte_count) + данные
unit_id = 1
function_code = 16  # Write Multiple Registers
start_addr = 10

# Упаковываем заголовок
data = struct.pack('>HHHBBHHB',
    transaction_id, protocol_id, length,
    unit_id, function_code, start_addr, quantity, byte_count)
# Добавляем значения (каждое как unsigned short big-endian)
for v in values:
    data += struct.pack('>H', v)

sock.send(data)
response = sock.recv(1024)
print('Ответ:', response.hex())
sock.close()

Сервер принял полезную нагрузку мгновенно, подтвердив, что сложные состояния можно переписать за одну транзакцию.

Фаза 3: Фингерпринтинг возможностей

Я написал цикл автоматического сканирования команд от 0x01 до 0x7F, чтобы составить карту функций OpenPLC без документации:

import socket
import struct

def test_function_code(ip, port, func):
    """Проверяет поддержку function code на устройстве"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(2)
    try:
        sock.connect((ip, port))
        # Минимальный запрос: читаем 1 регистр с адреса 0 (если func не поддерживается, вернётся исключение)
        request = struct.pack('>HHHBBHH', 1, 0, 6, 1, func, 0, 1)
        sock.send(request)
        response = sock.recv(1024)
        if len(response) >= 9:
            # Проверяем, что код функции в ответе совпадает с запросом (нет exception)
            if response[7] == func:
                return True
            # Если ответ содержит exception, код функции будет func | 0x80
            elif response[7] == (func | 0x80):
                return False
        return False
    except:
        return False
    finally:
        sock.close()

supported = []
for fc in range(1, 0x80):
    if test_function_code('192.168.1.100', 502, fc):
        supported.append(fc)
        print(f'[+] FC{fc:02X} поддерживается')

print('Поддерживаемые коды:', [hex(x) for x in supported])

Результат сканирования:

[+] FC01 Supported (Read Coils)
[+] FC02 Supported (Read Discrete Inputs)
[+] FC03 Supported (Read Holding Registers)
[+] FC04 Supported (Read Input Registers)
[+] FC05 Supported (Write Single Coil)
[+] FC06 Supported (Write Single Register)
[+] FC15 Supported (Write Multiple Coils)
[+] FC16 Supported (Write Multiple Registers)

Неподдерживаемые коды отклонялись с Exception 01 (Illegal Function).

Попытка получить метаданные (FC43)

Попробовал Function Code 43 (Modbus Encapsulated Interface Type 0x0E) для чтения тегов вендора и прошивки. OpenPLC отклонил пакет Illegal Function, защитив метаданные.

Фаза 4: Фаззинг, исключения и границы памяти

Устойчивость к исключениям

Я намеренно отправлял искажённые кадры: неверную длину, адреса вне диапазона, невалидные функции. ПЛК корректно генерировал исключения по спецификации:

01 — Illegal Function: команды вне диапазона.
02 — Illegal Data Address: запрос в мёртвую зону.
03 — Illegal Data Value: некорректные параметры.

Примечательно, что демон gracefully обрабатывал ошибки — ни падения, ни зависания.

Автоматическое определение границ памяти

С помощью бинарного поиска я нашёл точный потолок памяти Holding Registers:

import socket
import struct

def read_register(ip, port, addr):
    """Читает один holding register по адресу, возвращает значение или None при ошибке"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(2)
    try:
        sock.connect((ip, port))
        request = struct.pack('>HHHBBHH', 1, 0, 6, 1, 3, addr, 1)
        sock.send(request)
        response = sock.recv(1024)
        if len(response) >= 9 and response[7] == 3:
            # Извлекаем 2 байта данных
            value = struct.unpack('>H', response[9:11])[0]
            return value
        else:
            return None
    except:
        return None
    finally:
        sock.close()

def find_boundary(ip, port, low=0, high=65535):
    """Бинарный поиск последнего доступного адреса"""
    while low <= high:
        mid = (low + high) // 2
        val = read_register(ip, port, mid)
        if val is not None:
            # Проверяем следующий адрес
            next_val = read_register(ip, port, mid+1)
            if next_val is None:
                return mid
            else:
                low = mid + 1
        else:
            high = mid - 1
    return -1

boundary = find_boundary('192.168.1.100', 502)
print(f'Максимальный адрес: {boundary}')  # 8191

Скрипт сошёлся на смещении 8191 — это 8192 адреса (16 КБ RAM). Попытка прочитать 8192 вызывала Illegal Data Address.

Неожиданная особенность реализации

Во время фаззинга дискретных выходов (FC05) я заметил отклонение от спецификации. Официально для изменения состояния катушки разрешены только 0xFF00 (ON) и 0x0000 (OFF). Однако OpenPLC принимал и зеркалировал любые ненулевые значения (0x0001, 0x1234, 0xFFFF), трактуя их как TRUE. Именно такие находки окупают ручное исследование.

Системная реальность безопасности

Главный вывод — не программная ошибка, а фундаментальная уязвимость протокола:

Аутентификация: отсутствует
Авторизация: отсутствует
Шифрование: отсутствует

Если устройство доступно по TCP/502, любой может читать, писать и фаззить данные. Legacy-протоколы созданы для производительности и совместимости, поэтому безопасность должна обеспечиваться извне: сегментация сети, белые списки, активное логирование.

Обнаружение атак с высокой точностью

Действия атакующего оставляют характерные сетевые артефакты. Blue team может построить правила детектирования:

Сканирование кодов функций: мониторинг последовательных запросов разных FC от одного клиента.
Профилирование границ: паттерны бинарного поиска с последующим шквалом Exception 02.
Шторм исключений: резкий рост кодов 01, 02, 03 от одного узла — признак фаззера.
Асимметричные записи: белый список для FC05/06/15/16 только с HMI и инженерных терминалов.

Практический вывод

Если вы серьёзно настроены понять OT/ICS безопасность — отключите библиотеки, соберите маленький симулятор, откройте raw-сокет и напишите пакеты сами. Провода не врут.

Полный код инструментов (сборка пакетов, индикаторы аномалий, руководство по воспроизведению) доступен в открытом репозитории: Industrial Protocol Labs.

Пишем Modbus TCP с нуля: raw-сокеты и реверс-инжиниринг