Как строить сканирование активов в MaxPatrol VM. Ч.5 — Автоматизация по API. Ч.1 — IP-план, задачи сканирования

Моя цель - предложение широкого ассортимента товаров и услуг на постоянно высоком качестве обслуживания по самым выгодным ценам.

Добрый день, дорогие читатели!

Мы переходим к использованию API MaxPatrol VM по-взрослому.

В прошлой статье мы успешно авторизовались в MPVM, получили нужные для любого API-запроса headers.

Сегодня мы загрузим в MPVM IP-план Организации, создадим задачи по сканированию Пентестом и затронем задачи по Аудиту.

Примеры будут также на Python.

3 закона робототехники расписания задач

Мы с Вами в ч.2 и ч.3 разово просканировали бОльшую часть инфраструктуры Организации, узнали масштабы будущих работ по устранению уязвимостей. Возможно, что вручную отправили первые задачи по Устранению. Теперь надо поставить сканирование на поток, чтобы банально отслеживать изменения в лучшую или худшую сторону.

Расписание сканирований должно быть разработано совместно с ИТ и согласовано с ИТ.

Я рекомендую HostDiscovery выполнять не реже раза в сутки. С Pentest универсальную рекомендацию выдать сложно, т.к. необходимо учитывать объем инфраструктуры, количество коллекторов сканирования, пожелания по окнам сканирования от ИТ и еще множество других локальных факторов.

Фантаст Айзек Азимов в рассказе «Хоровод» 1942 года сформулировал 3 закона роботехники, определяющие этические принципы поведения роботов. Многие о них слышали, но я на всякий случай процитирую:

  1. Робот не может причинить вред человеку или своим бездействием допустить, чтобы человеку был причинен вред (безопасность человека).

  2. Робот обязан выполнять команды человека, если они не противоречат Первому закону (подчинение приказам).

  3. Робот обязан делать все необходимое для обеспечения своей безопасности, при условии, что это не противоречит Первому или Второму законам (собственная безопасность).

Я попытаюсь адаптировать эти законы под Управление уязвимостями в части проведения сканирований.

1. Сканер не может причинить вред инфраструктуре или отсутствием сканирования допустить, чтобы инфраструктуре был причинен вред.

Когда Вы согласуете регулярные сканирования, то первое время все последующие инциденты ИТ будут вешать на Вас. Это неизбежно. Новая игрушка притягивает наибольшее внимание в детском садике. А все мы знаем, что мальчики начинают взрослеть после 40, и то не всегда. Придётся регулярно показывать ИТ, согласованное с ИТ же расписание, в попытке доказать, что сканер даже рядом не пробегал во время падения сервиса.

В моей практике случаев, когда была нарушена рабоспособность сервиса, но сам сервис уже и так не был «поломан» = 0. Как правило, если сервис падает, то он разваливается от всего подряд, а не только от сканирования. Вспоминаем аналогию с игрушкой в детском садике. Чиним сервис -> сканы его больше не роняют. Готовьтесь защищаться и читать логи ОС/сервиса. Хоть это и не Ваша работа искать истинные причины неработоспособности, но надо это только Вам (если не хотите остаться крайним).

Но и внеплановые сканирования тоже никто не отменял. Не редки случаи, когда требуется внеплановый скан: процесс приемки ИС в эксплуатацию, вывод нового сервиса на периметр, контроль устранения конкретной CVE и еще великое множество всего, на что хватает процессной фантазии внутри Организации.

2. VM-щик должен повиноваться всем приказам, которые даёт IT-шник при согласовании расписания, кроме тех случаев, когда эти приказы противоречат Первому закону.

Если ИТ говорят, что эту систему в рабочее время/в будний день и т.д. сканировать нельзя, то спорить тут не о чем. Выбираем слот, который в первую очередь устроит ИТ. Но и не стоит забывать, что вариант не сканировать вовсе - не вариант.

3. VM-щик должен заботиться о безопасности своей Организации в той мере, в которой это не противоречит Первому и Второму законам.

Да, список исключений из сканирования неизбежно появится (по различным причинам). Задача «собственной безопасности» VM-щика - это жестко контролировать вход и выход ИТ-активов из списка исключений. Исключения из сканирования - это временная мера. Поговорка «нет ничего более постоянного, чем временное» не должна применяться к сканированию инфраструктуры.

IP-план

IP-план есть в каждой Организации. Но не каждая хранит его в CMDB или в IPAM. Чаще всего это экселька, в который множество неточностей, ошибок и опечаток. Да и актуальность Excel-табличек всегда вызывает вопросы.

Задача VM-щика любыми способами заставить предоставить ему актуальный IP-план в человекочитаемом (без опечаток), а лучше в машиночитаемом формате.

Все аргументы из серии «У Вашего коллеги есть учетка на коммутаторах (не факт, что на всех), пусть он вам VLAN-ы покажет» должны отсекаться уже на предлоге «У». Максимально полная и актуальная информация - минимально необходимое требование.

Здесь мы обсудим базовый пример IP-плана в машиночитаемом формате, который мы передадим в MPVM. Для нашего примера будем считать, что из IP-плана можно получить следующую информацию:

  • Физическое расположение сети (Офис, ЦОД, облако и т.п.).

  • Подсеть.

  • Маска сети.

  • VLAN.

  • Описание подсети.

  • Принадлежность подсети к DMZ или КСПД-сегменту.

В MPVM у нас созданы профили Pentest и HostDiscovery (либо используем стандартные). И создана статическая группа в MPVM, в которую мы перенесем полученный IP-план.

Нам нужно захардкодить id профилей и родительской статической группы. Проще и быстрее всего скопировать их из URL.

Пример для профиля:

Пример для группы:

HD = "13c32aa6-3a82-4e10-b709-941f39838aae"  # id профиля HostDiscovery в MPVM
Pentest = "b203b724-bc55-47d1-91d8-afbae230b19d"  # id профиля Pentest в MPVM
parentId = "1ed8aa9c-56c0-0001-0000-000000000006"  # id корневой статической группы в MPVM. Внутри неё будет воссоздана структура IP-плана

habr = {
    "segments": [
        {
            "name": "Habr",  # Физическое расположение сети (Офис, ЦОД, облако и т.п.).
            "parentId": parentId,  #
            "groupType": "static",  # Создаем статическую группу, в неё вложим динамические
            "networks": [ # Сеть, которой принадлежат DMZ и КСПД подсети. Всю её пентестом не проверяем, т.к. используется не весь набор доступных подсетей, но проверяем через HostDiscovery
                {
                    "groupType": "dynamic",  # Создаем динамическую группу по PDQL-запросу
                    "network": "192.168.0.0",  # Подсеть
                    "subnet": "16",  # Маска подсети
                    "vlan": "",  # Номер VLAN
                    "description": "Вся наша сеть",  # Описание
                    "scantype": "HD",  # Тип создаваемой задачи. Нужен для наименования
                    "scanprofile": HD,  # id профиля в задачи сканирования
                },
            ],
            "kspd": [ # Сети, определенные как КСПД
                {
                    "name": "КСПД",
                    "groupType": "static",  # Создаем статическую группу, в неё вложим динамические
                    "networks": [
                        {
                            "groupType": "dynamic",  # Создаем динамическую группу по PDQL-запросу
                            "network": "192.168.50.0",  # Подсеть
                            "subnet": "24",  # Маска подсети
                            "vlan": "50",  # Номер VLAN
                            "description": "Подсеть в КСПД",  # Описание
                            "scantype": "Pentest",  # Тип создаваемой задачи. Нужен для наименования
                            "scanprofile": Pentest,  # id профиля в задачи сканирования
                        },
                    ],
                }
            ],
            "dmz": [ # Сети, определенные как DMZ
                {
                    "name": "DMZ",
                    "groupType": "static",  # Создаем статическую группу, в неё вложим динамические
                    "networks": [  # Перечисление подсетей
                        {
                            "groupType": "dynamic",  # Создаем динамическую группу по PDQL-запросу
                            "network": "192.168.42.0",  # Подсеть
                            "subnet": "25",  # Маска подсети
                            "vlan": "42",  # Номер VLAN
                            "description": "Подсеть в DMZ",  # Описание
                            "scantype": "Pentest",  # Тип создаваемой задачи. Нужен для наименования
                            "scanprofile": Pentest,  # id профиля в задачи сканирования
                        },
                    ],
                }
            ],
        },
    ]
}

Переносим IP-план в MPVM

Переходим к процессу перекладки json-ов по API в MPVM.

Первое, что нам нужно, - это headers, которые мы получили в результате манипуляций с API из ч.0.

При создании динамических групп по нашим подсетям нам нужно провести проверку существования динамической группы с таким названием (в будущем нам это пригодится, когда будем в MPVM переносить состав информационных систем). Для этого нам нужно получить дерево/иерархию существующих групп в MPVM - через соответствующий API-запрос.

import requests
from json import loads
from ... import mpvm_base_url, MPVM_HTTPS_VERIFY # импорт из файла с глобальными параметрами: путь к полной цепочке сертификации в формате .pem и URL


def get_groups_tree(headers):

    url = f"{mpvm_base_url}/api/assets_temporal_readmodel/v2/groups/hierarchy"
    responce = requests.get(url=url, headers=headers, verify=MPVM_HTTPS_VERIFY)

    tree = loads(responce.text)

    return tree[0]

Иерархию групп получили. Теперь нам для API-запроса создания группы, нужно подготовить полезную нагрузку (payload). Payload для создания статической и динамической группы отличаются.

def make_static_payload(obj, parentId):
    payload = {  # Полезная нагрузка для создания статической (корневой) группы
        "name": obj["name"],  # Имя группы
        "parentId": parentId,  # id корневой группы
        "groupType": obj[
            "groupType"
        ],  # Тип создаваемой группы: статическая/динамическая
        "metadata": [],  # Метаданные
    }

    return payload


def make_net_dynamic_payload(obj, parentId):

    group_name = ""  # Имя дин. группы CIDR + VLAN (если есть) + Описание
    network = f'{obj["network"]}/{obj["subnet"]}'
    if obj["vlan"] != "":
        group_name += f"VLAN {obj['vlan']} "
    group_name += f'{network} {obj["description"]}'
    payload = {  # Полезная нагрузка для создания дин. группы
        "name": group_name,  # Имя группы
        "parentId": parentId,  # id корневой группы
        "groupType": obj[
            "groupType"
        ],  # Тип создаваемой группы: статическая/динамическая
        "organizationInfrastructure": {
            "usedNetworks": network
        },  # Заполняем используемые сети
        "predicate": f"Host.IpAddress in {network}",  # pdql-запрос дин. группы
        "metadata": [],  # Метаданные
    }

    logging.info(f"Start creating new MPVM asset group with name = {group_name}")

    return payload, group_name

Начинаем парсить наш IP-план.

import time

for segment in habr["segments"]:
    payload = make_static_payload(obj=segment, parentId=segment["parentId"])
    segment_parentId, flag = create_mpvm_group(
        tree=group_tree, headers=headers, payload=payload
    )
    for net in segment["habr"]:
        payload, group_name = make_net_dynamic_payload(
            obj=net, parentId=segment_parentId
        )
        flag = ""  # Флаг нужно ли создавать к дин. группе задачу по сканированию
        new_parentId, flag = create_mpvm_group(
            tree=group_tree, headers=headers, payload=payload
        )
        # Создание задачи по сканированию подсети
        if flag == "scan":
            scan_task_id = create_mpvm_scantask(
                headers=headers,
                obj=net,
                group_name=group_name,
                segment_name=segment["name"],
            )
        else:
    time.sleep(5) # Задержка 5 секунд для создания статической группы, прежде чем создавать в ней динамическую (на всякий случай)
    for kspd in segment["kspd"]:
        payload = make_static_payload(obj=kspd, parentId=segment_parentId)
        flag = ""
        kspd_parentId, flag = create_mpvm_group(
            tree=group_tree, headers=headers, payload=payload
        )
        for net in kspd["habr"]:
            payload, group_name = make_net_dynamic_payload(net, kspd_parentId)
            flag = ""
            new_parentId, flag = create_mpvm_group(
                tree=group_tree, headers=headers, payload=payload
            )
            # Создание задачи по сканированию подсети
            if flag == "scan":
                scan_task_id = create_mpvm_scantask(
                    headers=headers,
                    obj=net,
                    group_name=group_name,
                    segment_name=segment["name"],
                )
            else:
    for dmz in segment["dmz"]:
        payload = make_static_payload(obj=dmz, parentId=segment_parentId)
        flag = ""
        dmz_parentId, flag = create_mpvm_group(
            tree=group_tree, headers=headers, payload=payload
        )
        for net in dmz["habr"]:
            payload, group_name = make_net_dynamic_payload(net, dmz_parentId)
            flag = ""
            new_parentId, flag = create_mpvm_group(
                tree=group_tree, headers=headers, payload=payload
            )
            # Создание задачи по сканированию подсети
            if flag == "scan":
                scan_task_id = create_mpvm_scantask(
                    headers=headers,
                    obj=net,
                    group_name=group_name,
                    segment_name=segment["name"],
                )

В коде выше у нас осталась не описана функция create_mpvm_group. Идем детально ее смотреть.

def create_mpvm_group(tree, headers, payload):
    logging.info("Start module create_mpvm_group()")

    # Проверка существования создаваемой группы
    check_group_name_in_tree = check_group_in_mpvm(
        tree=tree, parentId=payload["parentId"], name=payload["name"]
    )
    new_groupId = ""
    flag = ""

    if check_group_name_in_tree == None:  # Если такой группы нет, то создаем новую
        new_groupId = create_new_mpvm_asset_group(headers=headers, payload=payload)
        if payload["groupType"] == "dynamic":
            flag = "scan"

    else:  # Если такая группа уже создана, то обновляем ее pdql-запрос
        new_groupId = check_group_name_in_tree
        if payload["groupType"] == "dynamic":
            update_mpvm_asset_group(
                headers=headers, payload=payload, group_id=check_group_name_in_tree
            )

    return new_groupId, flag

Ох, здесь у нас аж 3 больших пробела:

  • check_group_in_mpvm

  • create_new_mpvm_asset_group

  • update_mpvm_asset_group

Идём сверху вниз и начинаем с check_group_in_mpvm - эта функция проверяет, существует ли требуемая группа в иерархии групп MPVM. Здесь нам на помощь приходит рекурсия (наконец-то пригодились эти ваши алгоритмы из универа).

def check_group_in_mpvm(tree, parentId, name):

    children = search_mother_group_in_tree(tree=tree, parentId=parentId)

    id = search_asset_mpvm_in_tree(children=children, name=name)

    return id


# Во всей иерархии группы в MPVM ищем поддерево по id
def search_mother_group_in_tree(tree, parentId):
    if tree["id"] == parentId:
        return tree

    for i in range(len(tree["children"])):
        if tree["children"][i] != []:
            children = search_mother_group_in_tree(
                tree=tree["children"][i], parentId=parentId
            )
            if children != None:
                return children


# Ищем конкретную группу по имени внутри поддерева 
def search_asset_mpvm_in_tree(children, name):
    if children == None:
        return
    if children["name"] == name:
        return children["id"]

    for i in range(len(children["children"])):
        if children["children"][i] != []:
            id = search_asset_mpvm_in_tree(children=children["children"][i], name=name)
            if id != None:
                return id

Проверили, такой группы в MPVM нет. Создаем её функцией create_new_mpvm_asset_group. Для корректного создания группы нам нужно её создать и проверить ее создание.

def create_new_mpvm_asset_group(headers, payload):
    body = dumps(dict(payload), ensure_ascii=False).encode("utf-8")

    url = mpvm_base_url + "/api/assets_processing/v2/groups"
    responce = requests.post(
        url=url, headers=headers, data=body, verify=MPVM_HTTPS_VERIFY
    )

    create_task_id = loads(responce.text)[
        "operationId"
    ] 

    time.sleep(5)

    url = f"{mpvm_base_url}/api/assets_processing/v2/groups/operations/{create_task_id}"
    responce = requests.get(url=url, headers=headers, verify=MPVM_HTTPS_VERIFY)

    new_groupId = loads(responce.text)

    return new_groupId

Проверили, такая группа в MPVM есть. Тогда обновляем её pdql-запрос в функции update_mpvm_asset_group.

Данный API-метод не описан в оф. документации. Есть только в Postman-коллекции.

def update_mpvm_asset_group(headers, payload, group_id):

    url = f"{mpvm_base_url}/api/assets_processing/v2/groups/{group_id}"

    update_payload = {
        "type": "SetPredicateGroupCommand",
        "value": payload["predicate"],
    }

    body = f"[{update_payload}]"
    responce = requests.put(
        url=url, headers=headers, data=body, verify=MPVM_HTTPS_VERIFY
    )

    return

Создаем задачи по Пентесту подсетей

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

def create_mpvm_scantask(headers, obj, group_name, segment_name):
    network = f'{obj["network"]}/{obj["subnet"]}'

    payload = {
        "name": f'{segment_name} {obj["scantype"]} {group_name}',
        "scope": scopeid,
        "profile": obj["scanprofile"],
        "include": {
            "assets": [],
            "targets": [network],
            "assetsGroups": [],
        },
        "exclude": {"assets": [], "targets": [], "assetsGroups": []},
        "hostDiscovery": {"enabled": "false", "profile": "null"},
        "triggerParameters": {
            "type": "Daily",
            "fromDate": "",
            "atTime": "11:00:00",
            "timeZone": "+03:00",
            "daysOfWeek": [
                "monday",
                "tuesday",
                "wednesday",
                "thursday",
                "friday",
                "saturday",
                "sunday",
            ],
            "toDate": "",
            "isEnabled": "false",
        },
    }

    url = mpvm_base_url + "/api/scanning/v4/scanner_tasks/create"
    body = dumps(dict(payload), ensure_ascii=False).encode("utf-8")

    responce = requests.post(
        url=url, headers=headers, data=body, verify=MPVM_HTTPS_VERIFY
    )

    if responce.status_code == 200 or responce.status_code == 201:
        new_task_id = loads(responce.text)["id"]

        return new_task_id

    else:
        return

Как можно заметить, в payload есть параметры (в том числе обязательные), которые не описаны в документации к API. Как же быть в условиях недостаточности документации? Дарю свой личный лайфхак. Чтобы получить желаемые параметры, нужно создать руками в MPVM задачу с требуемыми параметрами, а затем выполнить API-метод запроса отчета по её id, который вернет вам правильный payload. Пример:

{
    "id": "...",
    "name": "...",
    "description": "",
    "include": {
        "assets": [],
        "targets": [],
        "assetsGroups": []
    },
    "exclude": {
        "assets": [],
        "targets": [],
        "assetsGroups": []
    },
    "isFqdnPriority": false,
    "agents": {
        "agents": [],
        "components": {
            "siemIds": null,
            "useCoreAgents": null
        }
    },
    "module": {
        "id": "...",
        "name": "..."
    },
    "scope": {
        "id": "...",
        "name": ".."
    },
    "profile": {
        "id": "...",
        "name": "..."
    },
    "completed": "...",
    "status": "finished",
    "created": "...",
    "lastRun": "...",
    "hasBookmarks": false,
    "lastRunErrorLevel": "green",
    "validationState": "valid",
    "hostDiscovery": {
        "enabled": false,
        "profile": null
    },
    "credentials": [],
    "triggerParameters": {
        "atTime": "11:00:00",
        "timeZone": "+03:00",
        "daysOfWeek": [
            "monday",
            "tuesday",
            "wednesday",
            "thursday",
            "friday"
        ],
        "type": "Daily",
        "fromDate": "...",
        "toDate": null,
        "isEnabled": true
    },
    "overrides": {},
    "deniedScanSettings": {
        "isEnabled": false,
        "periods": []
    },
    "groups": [
        {
            "id": "...",
            "name": "..."
        }
    ],
    "lastRunMode": "default",
    "connectionCheckSupported": false,
    "output": "assets"
}

Почему нельзя описать своё же API? - Я задаю этот риторический вопрос каждый раз, как открываю документацию к API. А об swagger остается только мечтать.

В автоматизацию я намеренно не закладываю настройку расписания, т.к. IP-план мы получаем автоматически, но вручную необходимо согласовать с ИТ слот сканирования. И новая задача намеренно попадает в группу «Задачи без групп», чтобы её можно было быстро обнаружить.

Объединяя все вышеописанные блоки кода, мы получаем вот такую иерархию групп, созданную в MPVM:

Так выглядит созданная дин. группа:

И вот такой список задач по сканированию:

Нас можно поздравить - мы успешно перенесли наш демонстрационный IP-план в MPVM.

Задачи Аудита

Автоматизацию задач Аудита можно сделать аналогично задачам по Пентесту подсетей. Нужно только нужный профиль сканирования подставить. УЗ проще вложить в кастомный профиль, чем прописывать скриптом. В качестве целей задачи указать id дин. групп. Я не автоматизировал Аудит по причине, что таких задач не так много, они редко добавляются, накладываются те же организационные ограничения и придётся вручную актуализировать список задач в коде.

Аудит АРМов

Отдельно поделюсь своими рекомендациями по Аудиту АРМов. Забудьте всё то, что написано у Вас в регламентах. Если по документам у Вас сотрудники обязаны не выключать ПК после окончания рабочего дня - они выключат, если выключать - оставят включенными. Я рекомендую сканированием АРМов заниматься ежедневно в рабочее время, т.к. у них постоянно плавающий онлайн. Кто-то в отпуске, кто-то в командировке, на удаленке, выключил ПК и ушел на обед, отключил от сети, перезагрузил АРМ и остальные человеческие факторы нормальной работы, прерывающие сканирование. Победить это невозможно, будем адаптироваться. Схема следующая:

  • Пентест АРМов, которые не сканировались Пентестом больше 4 дней раз в 3 часа с понедельника по пятницу. Первый старт в 9:00, следующий в 12:00, последний в 15:00. Пример PDQL-запроса для Windows АРМов:

WindowsHost.DomainRole = 'Domain Workstation' and (WindowsHost.@AuditTime < now() - 4d or WindowsHost.@AuditTime = null)
  • Аудит АРМов, которые не сканировались Аудитом больше 4 дней раз в 3 часа с понедельника по пятницу. Первый старт в 9:00, следующий в 12:00, последний в 15:00. Пример PDQL-запроса для Windows АРМов:

WindowsHost.DomainRole = 'Domain Workstation' and (WindowsHost.@PentestTime < now() - 4d or WindowsHost.@PentestTime = null)

Эта схема позволяет нам просканировать АРМ как минимум раз в неделю. Либо отследить, что он был онлайн, но сканирование не удалось. Эта схема не лишена недостатков, но об этом мы поговорим позже.

Источник: https://habr.com/ru/companies/gazpromcps/articles/1098568/


Интересные статьи

Интересные статьи

Мы писали на Хабре про нашу быструю и качественную библиотеку для простановки ударений и разрешения омографов на русском языке. Потом к ней доехали ещё акценторы для белорусского и украинского языков ...
В марте 2025 года «1С-Битрикс» выпустил обновление Виртуальной машины до версии 9.0.6, которое исправляет уязвимость, связанную с повышением привилегий bitrix->root (оценка CVSS: 8.0). «1С-Бит...
Те, кто следят за индустрией игр, наверняка заметили, что в списки анонсов среди наиболее ожидаемых игр все чаще стали попадаться так называемые соулслайки. Да, многие из них быстро уступают свое ...
Год назад у Битрикс24 появилась нативная интеграция с BI-сервисом DataLens: теперь пользователи CRM могут строить дашборд с бизнес-аналитикой за несколько кликов. Для создания таких аналитических отчё...
Предыстория Когда-то у меня возникла необходимость проверять наличие неотправленных сообщений в «1С-Битрикс: Управление сайтом» (далее Битрикс) и получать уведомления об этом. Пробле...