Строим лабораторию по исследованию ВПО

Ожидается, что мы сможем трассировать системные вызовы, дампировать сетевой трафик, а так же рассматривать открытый и закрытый исходный код сэмпла в поисках подозрительных паттернов.

Преимущества такой лаборатории видны сразу:

• существует мало современных SandBox решений, которые могут предоставить детализированный сырой лог сетевой и системной активности для Linux-платформ;

• обычно SandBox очень легко обходят, уводя в сон вредонос, проверяя локации, раскладки и прочие атрибуты. В данном решении вы сможете увидеть, спит ли сэмпл, а так же использовать свои ресурсы, чтобы увеличить время трассировки в отличии от стандартных 500 секунд песка.

• user execution очень органичен в стандартных решениях, необходимо использовать скрипты для выполнения дополнительных задач. В данном, ваша лаба полностью управляема вами.

• возможность имитировать ифнраструктуру вокруг сэмпла, часто может пригодиться для исследования инструментов.

Все круто, но не без недостатков:

• остается возможность прикрутить движки для вывода вердиктов, но лаба предназначена для ручного анализа;

• нужна другая архитектура для тестирования атак на пространство ядра, так как здесь ядро ОС общее;

• использование собственных ресурсов, вместо сервисных;

Архитектура представляет контур в рамках узла, где основные инструменты исследования будут развернуты в качестве драйвера pcap, бинарного файла strings, механизма eBPF и нормализатора на Python, а сэмпл будет воспроизводиться в контейнере.

Для изоляции двух сред будем использовать возможности Podman, который создаст нам виртуальное пространство с новыми namespace. Они должны ограничить возможность коммуникации, с другими обьектами. Seccomp запретит заражение пространства ядра, ограничив bpf, pivot_root, unshare, setns. Чтобы процесс не мог выйти из своего namespace или создать новый. Сgroup позволит выдать нужное количество ресурсов, а iptables разграничит по сети.

Для тех, кто не знаком с eBPF подкатом можно прочитать больше!

eBPF - это инструмент трассирования в Linux, который позволяет к точкам ОС крепить пользовательский код для мониторинга, фильтраций или изменения выполняемых функций. Работает он без системных вызовов поэтому зона его влияния - это функции разных пространств, сетевой трафик, а так же контроль точек LSM. Как упоминали раннее необходимо выбрать механизм и точку трассировки, такие как:

• uprobe / uretprobe - функции пользовательского пространства

• kprobe / kretprobe - функции пространства ядра

• tracepoint - точки для событий между пространствами (syscalls)

• network filter - точки сетевого трассирования

• LSM-hooks - хуки системы безопасности (Linux Security Modules)

Конфигурировать механизм можно используя системные вызовы bpf(), а так же же готовые инструменты bpftrace со скриптами bt. Схема работы такова:

Такие решения позволяют реализовать низкоуровневый функционал управления сетевым трафиком, запуском функций СЗИ, а так же становятся целью злоумышленников.

{
  "defaultAction": "SCMP_ACT_ALLOW",
  "architectures": [
    "SCMP_ARCH_X86_64",
    "SCMP_ARCH_X86",
    "SCMP_ARCH_X32"
  ],
  "syscalls": [
    {
      "names": [
        "acct",
        "add_key",
        "bpf",
        "clock_adjtime",
        "clock_settime",
        "create_module",
        "delete_module",
        "finit_module",
        "get_kernel_syms",
        "get_mempolicy",
        "init_module",
        "ioperm",
        "iopl",
        "kcmp",
        "kexec_file_load",
        "kexec_load",
        "keyctl",
        "lookup_dcookie",
        "mbind",
        "mount",
        "move_pages",
        "name_to_handle_at",
        "nfsservctl",
        "open_by_handle_at",
        "perf_event_open",
        "personality",
        "pivot_root",
        "process_vm_readv",
        "process_vm_writev",
        "ptrace",
        "query_module",
        "quotactl",
        "reboot",
        "request_key",
        "set_mempolicy",
        "setns",
        "settimeofday",
        "stime",
        "swapon",
        "swapoff",
        "sysfs",
        "syslog",
        "_sysctl",
        "umount",
        "umount2",
        "unshare",
        "uselib",
        "userfaultfd",
        "ustat",
        "vhangup"
      ],
      "action": "SCMP_ACT_ERRNO"
    }
  ]
}

  sudo podman run -d 
  --name malware-analysis-container 
  --network malware-net 
  --ip 172.20.0.100 
  --security-opt seccomp=./file.json 
  --memory=1024m --cpus=1 
  --systemd=always 
  localhost/fedora-systemd:latest /sbin/init

#!/bin/bash

IFACE="eth0" # Основной интерфейс в интернет
NETWORK_INTERFACE=$(sudo podman network inspect malware-net -f '{{(index .Plugins 0).InterfaceName}}')
DOCKER_IFACE=$NETWORK_INTERFACE
CONTAINER_IP="172.20.0.100"

# Сброс правил
iptables -F
iptables -t nat -F
iptables -X
iptables -t nat -X

# Политики по умолчанию
iptables -P INPUT DROP
iptables -P OUTPUT DROP
iptables -P FORWARD DROP

# Разрешаем localhost
iptables -A INPUT -i lo -j ACCEPT
iptables -A OUTPUT -o lo -j ACCEPT

# DMZ правила
iptables -A FORWARD -s $CONTAINER_IP -o $IFACE -j ACCEPT
iptables -A FORWARD -d $CONTAINER_IP -i $IFACE -m state --state ESTABLISHED,RELATED -j ACCEPT

# Запрет LAN
iptables -A FORWARD -s $CONTAINER_IP -d 192.168.0.0/16 -j DROP
iptables -A FORWARD -s $CONTAINER_IP -d 10.0.0.0/8 -j DROP
iptables -A FORWARD -s $CONTAINER_IP -d 172.16.0.0/12 -j DROP

echo "DMZ zone for $CONTAINER_IP has been configured!"

#define _GNU_SOURCE
#include <stdio.h>
#include <string.h>
#include <sys/utsname.h>
#include <dlfcn.h>

int (*orig_uname)(struct utsname *buf);

int uname(struct utsname *buf) {
    if (!orig_uname) {
        orig_uname = dlsym(RTLD_NEXT, "uname");
    }

    int ret = orig_uname(buf);

    strcpy(buf->sysname, "SuperSecureOS");
    strcpy(buf->nodename, "fakehost");
    strcpy(buf->release, "9.9.9");
    strcpy(buf->version, "v999");
    strcpy(buf->machine, "x86_999");
    strcpy(buf->domainname, "local.lan");

    return ret;
}

  sudo podman run -d 
  --security-opt label=disable 
  --name malware-analysis-container 
  --network malware-net 
  --ip 172.20.0.100 
  --security-opt seccomp=./file.json 
  --memory=1024m --cpus=1 
  --env LD_PRELOAD=/lib1/libfakeuname.so 
  --systemd=always 
  localhost/fedora-systemd:latest /sbin/init

/\* ===================== ПРОЦЕССЫ ===================== \*/
tracepoint:syscalls:sys_enter_execve
/ cgroup == $CGROUP_ID /
{
    printf("{\\"time\\":\\"%s\\",\\"probe\\":\\"%s\\",\\"pid\\":%d,\\"ppid\\":%d,\\"comm\\":\\"%s\\",\\"exec\\":\\"%s\\"}\\n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           curtask->real_parent->tgid,   // <-- ppid
           comm,
           str(args->filename));
}

#!/bin/bash
if [ $# -eq 0 ]; then
    echo "Usage: $0 <filename>"
    exit 1
fi

if [ ! -f "$1" ]; then
    echo "File $1 does not exist!"
    exit 1
fi

CONTAINER_PID=$(sudo podman inspect --format '{{.State.Pid}}' $(sudo podman ps -q | head -n 1))
CGROUP_PATH=$(cut -d: -f3 /proc/$CONTAINER_PID/cgroup)
CGROUP_ID=$(stat -c %i /sys/fs/cgroup$CGROUP_PATH)
echo "Tracing container with PID=$CONTAINER_PID, cgroup inode=$CGROUP_ID"


cat <<EOF > ./script.bt
#!/usr/bin/env bpftrace

/* ===================== ПРОЦЕССЫ ===================== */
tracepoint:syscalls:sys_enter_execve
/ cgroup == $CGROUP_ID /
{
    printf("{"time":"%s","probe":"%s","pid":%d,"ppid":%d,"comm":"%s","exec":"%s"}n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           curtask->real_parent->tgid,   // <-- ppid корректно
           comm,
           str(args->filename));
}

/* ===================== ФАЙЛЫ ===================== */
tracepoint:syscalls:sys_enter_openat
/ cgroup == $CGROUP_ID /
{
    printf("{"time":"%s","probe":"%s","pid":%d,"comm":"%s","pathname":"%s","flags":%d}n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           comm,
           str(args->filename),
           args->flags);
}

tracepoint:syscalls:sys_enter_unlinkat
/ cgroup == $CGROUP_ID /
{
    printf("{"time":"%s","probe":"%s","pid":%d,"comm":"%s","target":"%s"}n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           comm,
           str(args->pathname));
}

tracepoint:syscalls:sys_enter_renameat
/ cgroup == $CGROUP_ID /
{
    printf("{"time":"%s","probe":"%s","pid":%d,"comm":"%s","oldpath":"%s","newpath":"%s"}n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           comm,
           str(args->oldname),
           str(args->newname));
}

/* ===================== СЕТЬ ===================== */
tracepoint:syscalls:sys_enter_socket
/ cgroup == $CGROUP_ID /
{
    printf("{"time":"%s","probe":"%s","pid":%d,"comm":"%s","domain":%d,"type":%d,"protocol":%d}n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           comm,
           args->family,
           args->type,
           args->protocol);
}

tracepoint:syscalls:sys_enter_connect
/ cgroup == $CGROUP_ID /
{
    printf("{"time":"%s","probe":"%s","pid":%d,"comm":"%s","fd":%d,"addrlen":%d}n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           comm,
           args->fd,
           args->addrlen);
}

tracepoint:syscalls:sys_enter_accept
/ cgroup == $CGROUP_ID /
{
    printf("{"time":"%s","probe":"%s","pid":%d,"comm":"%s","fd":%d}n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           comm,
           args->fd);
}

/* ===================== UID/GID ===================== */
tracepoint:syscalls:sys_enter_setuid
/ cgroup == $CGROUP_ID /
{
    printf("{"time":"%s","probe":"%s","pid":%d,"comm":"%s","uid":%d}n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           comm,
           args->uid);
}

tracepoint:syscalls:sys_enter_setgid
/ cgroup == $CGROUP_ID /
{
    printf("{"time":"%s","probe":"%s","pid":%d,"comm":"%s","gid":%d}n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           comm,
           args->gid);
}

tracepoint:syscalls:sys_enter_setfsuid
/ cgroup == $CGROUP_ID /
{
    printf("{"time":"%s","probe":"%s","pid":%d,"comm":"%s","fsuid":%d}n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           comm,
           args->uid);
}

/* ===================== SECURITY-SENSITIVE ===================== */
tracepoint:syscalls:sys_enter_bpf
/ cgroup == $CGROUP_ID /
{
    printf("{"time":"%s","probe":"%s","pid":%d,"comm":"%s","cmd":%d}n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           comm,
           args->cmd);
}

tracepoint:syscalls:sys_enter_ptrace
/ cgroup == $CGROUP_ID /
{
    printf("{"time":"%s","probe":"%s","pid":%d,"comm":"%s","request":%d,"target":%d}n",
           strftime("%Y-%m-%d %H:%M:%S", nsecs),
           probe,
           pid,
           comm,
           args->request,
           args->pid);
}
EOF

echo "Start tracing..."
sudo bpftrace ./script.bt > ./syscall.txt &
echo "Start network dumping..."
NETWORK_INTERFACE=$(sudo podman network inspect malware-net -f '{{(index .Plugins 0).InterfaceName}}')
sudo tcpdump -i $NETWORK_ID > ./net.txt &
echo "Start strings search..."
sudo strings "$1" > ./strings.txt &

import json
import argparse
from datetime import datetime
import requests

def parse_syscalls(sys_file):
    events = []
    with open(sys_file) as f:
        for line in f:
            try:
                ev = json.loads(line)
                ev["source"] = "syscall"
                events.append(ev)
            except json.JSONDecodeError:
                events.append({"source": "syscall", "raw": line.strip(), "parse_error": True})
    return events

def parse_network(net_file, date=None):
    events = []
    with open(net_file) as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            time_str, rest = line.split(" ", 1)
            if date:
                ts = f"{date} {time_str}"
            else:
                ts = datetime.now().strftime("%Y-%m-%d ") + time_str
            events.append({
                "time": ts,
                "source": "net",
                "raw": line
            })
    return events

def parse_strings(strings_file, date=None):
    events = []
    with open(strings_file) as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            ts = date + " 00:00:00" if date else datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            events.append({
                "time": ts,
                "source": "strings",
                "raw": line
            })
    return events

def merge_events(*lists):
    all_events = []
    for lst in lists:
        all_events.extend(lst)
    # сортировка по времени
    def parse_time(ev):
        try:
            return datetime.strptime(ev["time"], "%Y-%m-%d %H:%M:%S.%f")
        except ValueError:
            try:
                return datetime.strptime(ev["time"], "%Y-%m-%d %H:%M:%S")
            except:
                return datetime.min
    return sorted(all_events, key=parse_time)

def send_to_ai(events, api_key):
    url = "https://gpt.serverspace.ru/v1/chat/completions"
    payload = {
        "model": "openai/gpt-4o",
        "max_tokens": 16384,
        "top_p": 0.1,
        "temperature": 0.6,
        "messages": [
            {
                "role": "user",
                "content": json.dumps(events)
            }
        ]
    }
    headers = {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}"
    }
    r = requests.post(url, headers=headers, data=json.dumps(payload))
    return r.json()


parser = argparse.ArgumentParser()
parser.add_argument("--sys", required=True, help="Syscall JSONL file")
parser.add_argument("--net", required=True, help="Network log txt file")
parser.add_argument("--strings", required=True, help="Strings output file")
parser.add_argument("--date", help="Date for network/strings events YYYY-MM-DD")
parser.add_argument("--out", help="Output JSONL file")
parser.add_argument("--api_key", help="GPT API key")
args = parser.parse_args()

sys_events = parse_syscalls(args.sys)
net_events = parse_network(args.net, date=args.date)
str_events = parse_strings(args.strings, date=args.date)

all_events = merge_events(sys_events, net_events)

if args.out:
    with open(args.out, "w") as f:
        for ev in all_events:
                f.write(json.dumps(ev) + "n")
else:
    for ev in all_events:
        print(json.dumps(ev))

if args.api_key:
    result = send_to_ai(all_events, args.api_key)
    print(json.dumps(result, indent=2))