Подробный гайд по аудиту SMB-трафика из PCAP через Python и Scapy - Часть 1

Подробный гайд по аудиту SMB-трафика из PCAP через Python и Scapy - Часть 1

Это взаимодополняющие подходы:

• PCAP/Scapy полезен, когда есть сетевой дамп и нужно понять, что происходило в SMB-трафике без доступа к хостам.
• Windows Event Log — более надежный источник для файлового аудита: он фиксирует доступ к шарам, имена файлов, пользователей, IP-адреса и причины отказа, даже если SMB-трафик шифруется.

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

---

---

---

1. Выбор подхода

Критерий	Аудит PCAP через Python/Scapy	SMB-аудит в Windows Event Log
Источник данных	Сетевой дамп	Логи сервера/рабочей станции
Видит доступ к файлам	Ограниченно, особенно при SMB3-шифровании	Да, если включен аудит
Видит пользователя	Часто нет/косвенно, зависит от протокола и шифрования	Да: пользователь, домен, тип входа
Видит имя файла	Возможно через SMB2-диссекторы, но сложно	Да, в событиях 5145/4663
Требует админ-доступ к хосту	Нет	Да
Сложность	Средняя/высокая из-за TCP-реассемблинга	Средняя
Надежность для расследования	Средняя	Высокая

---

Если задача — аудит доступа к файлам и шарам, лучше начинать с Windows Event Log. Если задача — разбор сетевого дампа, используйте Scapy или, что часто практичнее для SMB, tshark/pyshark.

---

2. Аудит PCAP-файлов через Python/Scapy

2.1. Установка окружения

Желательно использовать виртуальное окружение:

python -m venv venv

# Windows
venv\Scripts\Activate.ps1

# Linux/macOS
source venv/bin/activate

pip install scapy

Для некоторых сценариев полезны:

pip install pyshark

Но pyshark требует установленного tshark из состава Wireshark.

---

2.2. Базовый пример: найти SMB-пакеты

Простейший пример чтения PCAP и фильтрации портов SMB:

from scapy.all import rdpcap, TCP

SMB_PORTS = {445, 139}

packets = rdpcap("capture.pcap")

smb_packets = []

for pkt in packets:
    if TCP in pkt:
        if pkt[TCP].sport in SMB_PORTS or pkt[TCP].dport in SMB_PORTS:
            smb_packets.append(pkt)

print(f"Найдено пакетов с SMB-портами: {len(smb_packets)}")

Это только начало: порты 445/139 не гарантируют, что внутри полноценный SMB. Нужен разбор payload.

---

2.3. Важная особенность: SMB и TCP

SMB поверх TCP может быть:

• в одном пакете;
• разбит на несколько TCP-сегментов;
• содержать несколько SMB-сообщений в одном TCP-сегменте;
• быть составным, например SMB2 compound;
• быть зашифрованным при SMB3-encryption;
• использовать SMB Direct/RDMA, что плохо видно в обычном PCAP;
• использовать SMB over QUIC в новых сценариях.

Поэтому для серьезного анализа часто лучше подходят tshark/Wireshark. Scapy хорош для кастомной автоматизации, статистики и низкоуровневого разбора.

---

2.4. Полный скрипт: базовый парсер SMB1/SMB2 из PCAP

Скрипт:

• читает PCAP через Scapy;
• ищет TCP-порты 445/139;
• пытается найти SMB-сообщения внутри TCP payload;
• распознает:
  • SMB1: магический маркер 0xFF 'S' 'M' 'B';
  • SMB2: магический маркер 0xFE 'S' 'M' 'B';
  • SMB3 Transform/encrypted: магический маркер 0xFD 'S' 'M' 'B';
• извлекает команду, статус, Tree ID, Session ID, Message ID;
• пишет результат в CSV.

---

Сохраните как smb_pcap_parser.py:

#!/usr/bin/env python3

import argparse
import csv
import datetime
import struct
import sys
from pathlib import Path

from scapy.all import IP, IPv6, TCP, rdpcap, sniff

try:
    from scapy.sessions import TCPSession
except Exception:
    TCPSession = None


SMB2_COMMANDS = {
    0x0000: "NEGOTIATE",
    0x0001: "SESSION_SETUP",
    0x0002: "LOGOFF",
    0x0003: "TREE_CONNECT",
    0x0004: "TREE_DISCONNECT",
    0x0005: "CREATE",
    0x0006: "CLOSE",
    0x0007: "FLUSH",
    0x0008: "READ",
    0x0009: "WRITE",
    0x000A: "LOCK",
    0x000B: "IOCTL",
    0x000C: "CANCEL",
    0x000D: "ECHO",
    0x000E: "QUERY_DIRECTORY",
    0x000F: "CHANGE_NOTIFY",
    0x0010: "QUERY_INFO",
    0x0011: "SET_INFO",
    0x0012: "OPLOCK_BREAK",
}

SMB1_COMMANDS = {
    0x04: "CLOSE",
    0x25: "TRANS",
    0x2D: "OPEN_ANDX",
    0x2E: "READ_ANDX",
    0x2F: "WRITE_ANDX",
    0x32: "TRANS2",
    0x71: "TREE_DISCONNECT",
    0x72: "NEGOTIATE",
    0x73: "SESSION_SETUP_ANDX",
    0x74: "LOGOFF_ANDX",
    0x75: "TREE_CONNECT_ANDX",
    0xA2: "NT_CREATE_ANDX",
}

STATUS_NAMES = {
    0x00000000: "STATUS_SUCCESS",
    0xC0000005: "STATUS_ACCESS_VIOLATION",
    0xC000000D: "STATUS_INVALID_PARAMETER",
    0xC0000010: "STATUS_INVALID_DEVICE_REQUEST",
    0xC0000017: "STATUS_NO_MEMORY",
    0xC0000022: "STATUS_ACCESS_DENIED",
    0xC0000034: "STATUS_OBJECT_NAME_NOT_FOUND",
    0xC0000035: "STATUS_OBJECT_NAME_COLLISION",
    0xC000003A: "STATUS_OBJECT_PATH_NOT_FOUND",
    0xC0000043: "STATUS_SHARING_VIOLATION",
    0xC0000064: "STATUS_NO_SUCH_USER",
    0xC000006A: "STATUS_WRONG_PASSWORD",
    0xC000006D: "STATUS_LOGON_FAILURE",
    0xC000006E: "STATUS_ACCOUNT_RESTRICTION",
    0xC000006F: "STATUS_INVALID_LOGON_HOURS",
    0xC0000070: "STATUS_INVALID_WORKSTATION",
    0xC0000071: "STATUS_PASSWORD_EXPIRED",
    0xC0000072: "STATUS_ACCOUNT_DISABLED",
    0xC000009A: "STATUS_INSUFFICIENT_RESOURCES",
    0xC000015B: "STATUS_NOT_LOGGED_ON",
    0xC000018B: "STATUS_NO_SUCH_USER",
}

CSV_FIELDS = [
    "time_iso",
    "src_ip",
    "dst_ip",
    "smb",
    "command",
    "command_name",
    "status",
    "status_name",
    "tree_id",
    "session_id",
    "message_id",
    "note",
]


def iso_time(ts):
    return datetime.datetime.fromtimestamp(float(ts), tz=datetime.timezone.utc).isoformat()


def load_packets(path):
    """
    Пытаемся читать PCAP с TCP-сессией/реассемблингом, если доступно.
    Если не получается — читаем обычным rdpcap.
    """
    if TCPSession is not None:
        try:
            return sniff(offline=str(path), session=TCPSession)
        except Exception as e:
            print(f"TCPSession failed: {e}; falling back to rdpcap", file=sys.stderr)

    return rdpcap(str(path))


def get_ips(pkt):
    if IP in pkt:
        return pkt[IP].src, pkt[IP].dst
    if IPv6 in pkt:
        return pkt[IPv6].src, pkt[IPv6].dst
    return "", ""


def iter_smb_messages(payload):
    """
    Пытается выделить SMB-сообщения из TCP payload.

    Для SMB поверх TCP часто используется 4-байтовый транспортный заголовок:
    [тип: 1 байт][длина: 3 байта]

    Если сразу видим магический маркер SMB, считаем остаток одним SMB-блоком.
    """
    i = 0
    n = len(payload)

    while i < n:
        # Если сразу видим SMB magic, отдаем остаток как один блок.
        if payload[i:i + 4] in (b"\xffSMB", b"\xfeSMB", b"\xfdSMB"):
            yield payload[i:]
            return

        # Нужно минимум 4 байта под транспортный заголовок.
        if i + 4 > n:
            return

        msg_type = payload[i]

        # Для SMB-сообщений обычно ожидаем тип 0x00.
        if msg_type != 0x00:
            return

        length = int.from_bytes(payload[i + 1:i + 4], "big")
        start = i + 4
        end = start + length

        if length == 0:
            i += 4
            continue

        if end > n:
            # Частичное сообщение: отдаем то, что есть.
            yield payload[start:]
            return

        yield payload[start:end]
        i = end


def parse_smb2_chain(blob, ts, src, dst):
    """
    Разбор цепочки SMB2-заголовков внутри одного SMB-сообщения.
    """
    offset = 0

    while offset + 64 <= len(blob):
        if blob[offset:offset + 4] != b"\xfeSMB":
            break

        try:
            _structure_size = struct.unpack("<H", blob[offset + 4:offset + 6])[0]
            _credit_charge = struct.unpack("<H", blob[offset + 6:offset + 8])[0]
            status = struct.unpack("<I", blob[offset + 8:offset + 12])[0]
            command = struct.unpack("<H", blob[offset + 12:offset + 14])[0]
            _credit = struct.unpack("<H", blob[offset + 14:offset + 16])[0]
            flags = struct.unpack("<I", blob[offset + 16:offset + 20])[0]
            next_command = struct.unpack("<I", blob[offset + 20:offset + 24])[0]
            message_id = struct.unpack("<Q", blob[offset + 24:offset + 32])[0]

            # SMB2_FLAGS_ASYNC_COMMAND = 0x00000002
            if flags & 0x00000002:
                async_id = struct.unpack("<Q", blob[offset + 32:offset + 40])[0]
                tree_id = ""
                session_id = struct.unpack("<Q", blob[offset + 40:offset + 48])[0]
                note = f"async_id={async_id}"
            else:
                tree_id = struct.unpack("<I", blob[offset + 36:offset + 40])[0]
                session_id = struct.unpack("<Q", blob[offset + 40:offset + 48])[0]
                note = ""

        except struct.error:
            break

        yield {
            "time_iso": iso_time(ts),
            "src_ip": src,
            "dst_ip": dst,
            "smb": "SMB2",
            "command": command,
            "command_name": SMB2_COMMANDS.get(command, f"0x{command:04x}"),
            "status": status,
            "status_name": STATUS_NAMES.get(status, f"0x{status:08x}"),
            "tree_id": tree_id,
            "session_id": session_id,
            "message_id": message_id,
            "note": note,
        }

        if next_command == 0:
            break

        # Защита от некорректного значения.
        if next_command < 64:
            break

        offset += next_command


def parse_smb1(blob, ts, src, dst):
    """
    Минимальный разбор SMB1-заголовка.
    """
    if len(blob) < 32:
        return None

    try:
        command = blob[4]
        status = struct.unpack("<I", blob[5:9])[0]
        flags2 = struct.unpack("<H", blob[10:12])[0]
        tree_id = struct.unpack("<H", blob[24:26])[0]
        uid = struct.unpack("<H", blob[28:30])[0]
        mid = struct.unpack("<H", blob[30:32])[0]
    except struct.error:
        return None

    return {
        "time_iso": iso_time(ts),
        "src_ip": src,
        "dst_ip": dst,
        "smb": "SMB1",
        "command": command,
        "command_name": SMB1_COMMANDS.get(command, f"0x{command:02x}"),
        "status": status,
        "status_name": STATUS_NAMES.get(status, f"0x{status:08x}"),
        "tree_id": tree_id,
        "session_id": uid,
        "message_id": mid,
        "note": f"flags2=0x{flags2:04x}",
    }


def parse_packet(pkt, ports):
    """
    Разбор одного пакета.
    """
    if TCP not in pkt:
        return []

    tcp = pkt[TCP]

    if tcp.sport not in ports and tcp.dport not in ports:
        return []

    payload = bytes(tcp.payload)
    if not payload:
        return []

    src, dst = get_ips(pkt)
    ts = float(pkt.time)

    records = []

    for message in iter_smb_messages(payload):
        if not message:
            continue

        magic = message[:4]

        if magic == b"\xfeSMB":
            records.extend(parse_smb2_chain(message, ts, src, dst))

        elif magic == b"\xffSMB":
            rec = parse_smb1(message, ts, src, dst)
            if rec:
                records.append(rec)

        elif magic == b"\xfdSMB":
            records.append({
                "time_iso": iso_time(ts),
                "src_ip": src,
                "dst_ip": dst,
                "smb": "SMB3_TRANSFORM",
                "command": "",
                "command_name": "ENCRYPTED_TRANSFORM",
                "status": "",
                "status_name": "",
                "tree_id": "",
                "session_id": "",
                "message_id": "",
                "note": "SMB3 transform header; payload likely encrypted",
            })

    return records


def write_csv(rows, output_path):
    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=CSV_FIELDS)
        writer.writeheader()
        writer.writerows(rows)


def main():
    parser = argparse.ArgumentParser(description="Parse SMB traffic from PCAP using Scapy.")
    parser.add_argument("pcap", help="Path to PCAP file")
    parser.add_argument("-o", "--output", default="smb_events.csv", help="Output CSV file")
    parser.add_argument("--ports", default="445,139", help="TCP ports to inspect")
    args = parser.parse_args()

    ports = {int(p.strip()) for p in args.ports.split(",") if p.strip()}

    path = Path(args.pcap)
    if not path.exists():
        print(f"File not found: {path}", file=sys.stderr)
        sys.exit(1)

    print(f"Reading packets from {path} ...", file=sys.stderr)
    packets = load_packets(path)

    rows = []

    for i, pkt in enumerate(packets, start=1):
        rows.extend(parse_packet(pkt, ports))

        if i % 100000 == 0:
            print(f"Processed {i} packets...", file=sys.stderr)

    write_csv(rows, args.output)
    print(f"Wrote {len(rows)} SMB records to {args.output}")


if __name__ == "__main__":
    main()

---

---

Запуск:

---

---

python smb_pcap_parser.py capture.pcap -o smb_events.csv

---

---

Если нужно смотреть только порт 445:

python smb_pcap_parser.py capture.pcap --ports 445 -o smb_events.csv

---

2.5. Что будет в CSV

Пример полей:

Поле	Описание
time_iso	Время пакета в UTC
src_ip	Источник
dst_ip	Назначение
smb	SMB1 / SMB2 / SMB3_TRANSFORM
command	Числовой код команды
command_name	Имя команды: CREATE, READ, WRITE и т.д.
status	NTSTATUS
status_name	Расшифровка статуса
tree_id	Tree ID для SMB2 или TID для SMB1
session_id	Session ID для SMB2 или UID для SMB1
message_id	Message ID для SMB2 или MID для SMB1
note	Дополнительная информация

---

2.6. Что можно анализировать в таком CSV

Например:

• кто подключался к серверу;
• какие команды SMB выполнялись;
• были ли ошибки STATUS_ACCESS_DENIED;
• есть ли подозрительные TREE_CONNECT к ADMIN$, C$, IPC$;
• много ли SESSION_SETUP с ошибками;
• есть ли массовые CREATE/READ/WRITE;
• использовался ли SMB1.

---

Если установлен pandas, можно быстро анализировать:

import pandas as pd

df = pd.read_csv("smb_events.csv")

print(df.groupby(["src_ip", "command_name"]).size().sort_values(ascending=False).head(20))

print(df[df["status_name"] == "STATUS_ACCESS_DENIED"])

print(df[df["command_name"] == "TREE_CONNECT"])

---

2.7. Ограничения Scapy-подхода

1. Нет полноценной реассемблинга всех сложных случаев.

Для простых задач хватает TCPSession, но для сложных дампов лучше tshark.

2. SMB3-шифрование.

Если включено шифрование, вы можете видеть заголовки, но не сможете извлечь имена файлов и содержимое запросов.

3. Имена файлов в SMB2 CREATE не извлекаются простым парсером заголовков.

---

---

Для этого нужно полноценно разбирать тело SMB2 CREATE-запроса, кодировки, смещения, структуру запроса и ответов.

4. Большие PCAP могут обрабатываться медленно.

Scapy удобен, но не всегда самый быстрый.

5. SMB Direct/RDMA и SMB over QUIC.

Не покрываются простым TCP-парсером.

---

2.8. Когда лучше использовать tshark/pyshark

Если нужно быстро получить поля протокола, например имена файлов, лучше использовать tshark.

Примеры:

tshark -r capture.pcap -Y "tcp.port == 445" -z conv,tcp

---

Посмотреть SMB2-сообщения:

tshark -r capture.pcap -Y "smb2" -T fields \
  -e frame.time_epoch \
  -e ip.src \
  -e ip.dst \
  -e smb2.cmd \
  -e smb2.status

---

Попробовать получить имена файлов из SMB2 CREATE:

tshark -r capture.pcap -Y "smb2.cmd == 5" -T fields \
  -e frame.time_epoch \
  -e ip.src \
  -e ip.dst \
  -e smb2.filename

Поля могут отличаться в зависимости от версии Wireshark/tshark и типа SMB-сообщений.

---

Пример использования pyshark:

import pyshark

cap = pyshark.FileCapture(
    "capture.pcap",
    display_filter="smb2"
)

for pkt in cap:
    if hasattr(pkt, "smb2"):
        cmd = pkt.smb2.get_field("cmd")
        filename = pkt.smb2.get_field("filename")

        print(pkt.sniff_time, pkt.source, pkt.destination, cmd, filename)

---

Информация предоставлена в ознакомительных целях. Применение описанных настроек в системах, должно осуществляться только после согласования с ответственными за информационную безопасность и в соответствии с требованиями ФСТЭК, ФСБ и иных уполномоченных органов.

---