Оптимизиране на работата с Check Point API чрез Python SDK

Оптимизиране на работата с Check Point API чрез Python SDKЦялата мощ на взаимодействието с API се разкрива при съвместната работа с програмен код, когато се появят възможности динамично да се формулират API заявки и инструменти за анализ на API отговори. Все пак, малко забележимо остава Python Software Development Kit (наричан по-нататък — Python SDK) за Check Point Management API, и не напразно. Той значително улеснява живота на разработчиците и ентусиастите на автоматизацията. Python придоби огромна популярност в последно време и реших да запълня пропуските и да направя преглед на основните възможности на Check Point API Python Development Kit. Тази статия служи като отлично допълнение на друга статия в Хабра Check Point R80.10 API. Управление чрез CLI, скриптове и не само.. Ние ще разгледаме как да напишем скриптове, използвайки Python SDK, и ще се спрем подробно на новите функции на Management API в версия 1.6 (поддържа се от R80.40). За разбиране на статията ще са необходими основни познания по работа с API и Python.

Check Point активно развива API и в момента са се появили:

Python SDK към момента поддържа взаимодействие само с Management API и Gaia API. Ние ще разгледаме най-важните класове, методи и променливи в този модул.

Оптимизиране на работата с Check Point API чрез Python SDK

Инсталация на модула

Модулът cpapi се инсталира бързо и лесно от официалния репозитарий на Check Point в github чрез pip. Подробна инструкция за инсталация има в README.md. Този модул е адаптиран за работа с версии на Python 2.7 и 3.7. В тази статия примери ще бъдат представени с използване на Python 3.7. Въпреки това, Python SDK може да бъде стартиран директно от управляващия сървър на Check Point (Smart Management), но там се поддържа само версия на Python 2.7, затова в последния раздел ще бъде предоставен код за версия 2.7. Веднага след инсталацията на модула, препоръчвам да погледнете примерите в директориите examples_python2 и examples_python3.

Започване на работа

За да имаме възможност да работим с компонентите на модула cpapi, е необходимо да импортираме от модула cpapi поне най-малко два задължителни класа:

APIClient и APIClientArgs

от cpapi импортиране на APIClient, APIClientArgs

Клас APIClientArgs отговаря за параметрите за свързване с API сървъра, а класът APIClient отговаря за взаимодействието с API.

Определяме параметрите за свързване

За да определите различни параметри за свързване с API, трябва да създадете екземпляр на класа APIClientArgs. По принцип, параметрите му са предварително зададени и при стартиране на скрипта на управленския сървър не е необходимо да ги посочвате.

client_args = APIClientArgs()

Но при стартиране на външен хост, трябва да посочите поне IP адрес или име на хост на API сървъра (който е и управленският сървър). В примера по-долу определяме параметъра за свързване server и му присвояваме IP адреса на управленския сървър като низ.

client_args = APIClientArgs(server='192.168.47.241')

Нека разгледаме всички параметри и техните стойности по подразбиране, които могат да се използват при свързване с API сървъра:

Аргументите на метода __init__ на класа APIClientArgs

class APIClientArgs:
    """
    Този клас предоставя аргументи за конфигурация на APIClient.
    Всички аргументи са конфигурирани с техните стойности по подразбиране.
    """

    # портът е зададен на None по подразбиране, но се заменя с 443, ако не е указан
    # възможните стойности за контекста - web_api (по подразбиране) или gaia_api
    def __init__(self, port=None, fingerprint=None, sid=None, server="127.0.0.1", http_debug_level=0,
                 api_calls=None, debug_file="", proxy_host=None, proxy_port=8080,
                 api_version=None, unsafe=False, unsafe_auto_accept=False, context="web_api"):
        self.port = port
        # отпечатък на управленския сървър
        self.fingerprint = fingerprint
        # session-id.
        self.sid = sid
        # име или IP адрес на управленския сървър
        self.server = server
        # ниво на отстраняване на грешки
        self.http_debug_level = http_debug_level
        # масив с всички API извиквания (за целите на отстраняване на грешки)
        self.api_calls = api_calls if api_calls else []
        # име на файла за отстраняване на грешки. Ако е оставен празен, данните за отстраняване на грешки няма да бъдат записвани на диск.
        self.debug_file = debug_file
        # адрес на HTTP прокси сървър (без "http://")
        self.proxy_host = proxy_host
        # порт на HTTP прокси
        self.proxy_port = proxy_port
        # API версия на управленския сървър
        self.api_version = api_version
        # Указва, че клиентът не трябва да проверява сертификата на сървъра
        self.unsafe = unsafe
        # Указва, че клиентът автоматично трябва да приеме и запази сертификата на сървъра
        self.unsafe_auto_accept = unsafe_auto_accept
        # Контекст на използване на клиента - по подразбиране web_api
        self.context = context

Смятам, че аргументите, които могат да се използват в екземплярите на класа APIClientArgs, са интуитивно ясни за администраторите на Check Point и не се нуждаят от допълнителни коментари.

Свързваме се чрез APIClient и мениджъра на контекста

Клас APIClient най-удобно е да се използва чрез мениджъра на контекста. Всичко, което трябва да се предаде на екземпляра на класа APIClient, са параметрите за свързване, които бяха определени в предишната стъпка.

с APIClient(client_args) като client:

Мениджърът на контекста няма да извърши автоматично извикването на login към API сървъра, но ще извърши извикване на logout при излизане от него. Ако по някакви причини logout в края на работата с API извиквания не е необходим, трябва да започнете работа без използване на мениджър на контекста:

client = APIClient(clieng_args)

Проверка на свързаността

Най-лесно е да се провери дали свързаността отговаря на зададените параметри чрез метода check_fingerprint. Ако проверката на хеш сумата sha1 за цифровия отпечатък на сертификата на API сървъра не е успешна (методът е върнал False), обикновено това е предизвикано от проблеми със свързаността и можем да прекратим изпълнението на програмата (или да дадем на потребителя възможност да коригира данните за свързване):

    if client.check_fingerprint() is False:
        print("Не беше възможно да се получи цифровият отпечатък на сървъра - Проверете свързаността със сървъра.")
        exit(1)

Имайте предвид, че в бъдеще класът APIClient ще проверява при всяко API извикване (методите api_call и api_query, за които ще говорим по-нататък) sha1 цифровия отпечатък на сертификата на API сървера. А ако при проверката на sha1 цифровия отпечатък на сертификата на API сървера бъде открит проблем (сертификатът е непознат или е бил променен), методът check_fingerprint ще предостави възможност да добавите/промените информацията за него на локалната машина в автоматичен режим. Тази проверка може да бъде напълно деактивирана (но такова може да се препоръча само в случай на стартиране на скриптове на самия API сървър, при свързване към 127.0.0.1), използвайки аргумента APIClientArgs — unsafe_auto_accept (вижте повече за APIClientArgs по-рано в „Определяне на параметрите за свързване“).

client_args = APIClientArgs(unsafe_auto_accept=True)

Вход в API сървъра

У APIClient има цели 3 метода за вход в API сървъра, и всеки от тях запомня стойността sid(идентификатор на сесия), който автоматично се използва в всяко следващо API извикване в заглавката (името в заглавката на този параметър — X-chkp-sid), така че не е необходимо допълнително да обработвате този параметър.

Метод на вход

Вариант с използване на потребителско име и парола (в примера потребителското име admin и паролата 1q2w3e са предадени като позиционни аргументи):

     login = client.login('admin', '1q2w3e')  

В метода на вход са налични също и допълнителни опционални параметри, по-долу са имената им и стойностите по подразбиране:

continue_last_session=False, domain=None, read_only=False, payload=None

Метод login_with_api_key

Вариант с използване на API ключ (поддържа се от версията на управление R80.40/Management API v1.6, „3TsbPJ8ZKjaJGvFyoFqHFA==“ това е стойността на API ключа за един от потребителите на сървера за управление с метода на авторизация API ключ)

     login = client.login_with_api_key('3TsbPJ8ZKjaJGvFyoFqHFA==') 

В метода login_with_api_key са налични същите опционални параметри както и в метода login.

Методът login_as_root

Опция за вход на локалната машина с API сървера:

     login = client.login_as_root()

За този метод са налични само две опционални параметри:

domain=None, payload=None

И накрая самите API извиквания

Имаме два варианта за извикване на API чрез методи api_call и api_query. Нека разберем каква е разликата между тях.

api_call

Този метод е приложим за всякакви извиквания. Трябва да предадем последната част за api извикването и payload в тялото на заявката, когато е необходимо. Ако payload е празен, не е нужно да го предаваме изобщо:

api_versions = client.api_call('show-api-versions') 

Изходът за тази заявка под кат:

In [23]: api_versions                                                           
Out[23]: 
APIResponse({
    "data": {
        "current-version": "1.6",
        "supported-versions": [
            "1",
            "1.1",
            "1.2",
            "1.3",
            "1.4",
            "1.5",
            "1.6"
        ]
    },
    "res_obj": {
        "data": {
            "current-version": "1.6",
            "supported-versions": [
                "1",
                "1.1",
                "1.2",
                "1.3",
                "1.4",
                "1.5",
                "1.6"
            ]
        },
        "status_code": 200
    },
    "status_code": 200,
    "success": true
})
show_host = client.api_call('show-host', {'name' : 'h_8.8.8.8'})

Изходът за тази заявка под кат:

В [25]: show_host                                                              
Out[25]: 
APIResponse({
    "data": {
        "color": "черен",
        "comments": "",
        "domain": {
            "domain-type": "домен",
            "name": "SMC User",
            "uid": "41e821a0-3720-11e3-aa6e-0800200c9fde"
        },
        "groups": [],
        "icon": "Objects/host",
        "interfaces": [],
        "ipv4-address": "8.8.8.8",
        "meta-info": {
            "creation-time": {
                "iso-8601": "2020-05-01T21:49+0300",
                "posix": 1588358973517
            },
            "creator": "admin",
            "last-modifier": "admin",
            "last-modify-time": {
                "iso-8601": "2020-05-01T21:49+0300",
                "posix": 1588358973517
            },
            "lock": "разблокиран",
            "validation-state": "ок"
        },
        "name": "h_8.8.8.8",
        "nat-settings": {
            "auto-rule": false
        },
        "read-only": false,
        "tags": [],
        "type": "host",
        "uid": "c210af07-1939-49d3-a351-953a9c471d9e"
    },
    "res_obj": {
        "data": {
            "color": "черен",
            "comments": "",
            "domain": {
                "domain-type": "домен",
                "name": "SMC User",
                "uid": "41e821a0-3720-11e3-aa6e-0800200c9fde"
            },
            "groups": [],
            "icon": "Objects/host",
            "interfaces": [],
            "ipv4-address": "8.8.8.8",
            "meta-info": {
                "creation-time": {
                    "iso-8601": "2020-05-01T21:49+0300",
                    "posix": 1588358973517
                },
                "creator": "admin",
                "last-modifier": "admin",
                "last-modify-time": {
                    "iso-8601": "2020-05-01T21:49+0300",
                    "posix": 1588358973517
                },
                "lock": "разблокиран",
                "validation-state": "ок"
            },
            "name": "h_8.8.8.8",
            "nat-settings": {
                "auto-rule": false
            },
            "read-only": false,
            "tags": [],
            "type": "host",
            "uid": "c210af07-1939-49d3-a351-953a9c471d9e"
        },
        "status_code": 200
    },
    "status_code": 200,
    "success": true
})

api_query

Важно да подчертая, че този метод е приложим само за повиквания, при които изходът предполага offset (изместване). Такъв изход настъпва, когато в него има или може да съдържа голямо количество информация. Например, това може да бъде заявка за списък на всички създадени обекти от тип хост на управляващия сървър. За такива заявки API връща списък от 50 обекта по подразбиране (лимитът може да бъде увеличен до 500 обекта в отговора). И за да не се извиква информацията многократно, променяйки параметъра offset в API заявката, съществува метод api_query, който автоматизира тази работа. Примери за повиквания, където се изисква този метод: show-sessions, show-hosts, show-networks, show-wildcards, show-groups, show-address-ranges, show-simple-gateways, show-simple-clusters, show-access-roles, show-trusted-clients, show-packages. Всъщност, в името на тези API повиквания виждаме думи в множествено число, така че тези повиквания ще бъдат по-лесни за обработка чрез api_query

show_hosts = client.api_query('show-hosts') 

Изходът за тази заявка под кат:

В [21]: show_hosts                                                             
Out[21]: 
APIResponse({
    "data": [
        {
            "domain": {
                "domain-type": "domain",
                "name": "SMC User",
                "uid": "41e821a0-3720-11e3-aa6e-0800200c9fde"
            },
            "ipv4-address": "192.168.47.1",
            "name": "h_192.168.47.1",
            "type": "host",
            "uid": "5d7d7086-d70b-4995-971a-0583b15a2bfc"
        },
        {
            "domain": {
                "domain-type": "domain",
                "name": "SMC User",
                "uid": "41e821a0-3720-11e3-aa6e-0800200c9fde"
            },
            "ipv4-address": "8.8.8.8",
            "name": "h_8.8.8.8",
            "type": "host",
            "uid": "c210af07-1939-49d3-a351-953a9c471d9e"
        }
    ],
    "res_obj": {
        "data": {
            "from": 1,
            "objects": [
                {
                    "domain": {
                        "domain-type": "domain",
                        "name": "SMC User",
                        "uid": "41e821a0-3720-11e3-aa6e-0800200c9fde"
                    },
                    "ipv4-address": "192.168.47.1",
                    "name": "h_192.168.47.1",
                    "type": "host",
                    "uid": "5d7d7086-d70b-4995-971a-0583b15a2bfc"
                },
                {
                    "domain": {
                        "domain-type": "domain",
                        "name": "SMC User",
                        "uid": "41e821a0-3720-11e3-aa6e-0800200c9fde"
                    },
                    "ipv4-address": "8.8.8.8",
                    "name": "h_8.8.8.8",
                    "type": "host",
                    "uid": "c210af07-1939-49d3-a351-953a9c471d9e"
                }
            ],
            "to": 2,
            "total": 2
        },
        "status_code": 200
    },
    "status_code": 200,
    "success": true
})

Обработка на резултатите от API повикванията

След това можете да използвате променливите и методите на класа APIResponse(както вътре в контекстния мениджър, така и извън него). Класът APIResponse предлага 4 метода и 5 променливи, на най-важните ще се спрем по-подробно.

Оптимизиране на работата с Check Point API чрез Python SDK

success

За начало, би било добре да се уверим, че API повикването е преминало успешно и е върнало резултати. За това има метод success:

В [49]: api_versions.success                                                   
Изход[49]: True

Връща True, ако API повикването е преминало успешно (Код на отговора — 200) и False, ако не е успешно (всеки друг код на отговора). Удобно е да се използва веднага след API повикването, за да се изведе различна информация в зависимост от кода на отговора.

if api_ver.success: 
    print(api_versions.data) 
else: 
    print(api_versions.err_message) 

statuscode

Връща кода на отговора след изпълнение на API повикването.

В [62]: api_versions.status_code                                               
Изход[62]: 400

Възможни кодове на отговорите: 200,400,401,403,404,409,500,501.

set_success_status

В този случай може да се наложи да промените стойността на статуса success. Технически, можете да поставите всичко, дори обикновен текст. Но реален пример може да бъде нулирането на този параметър на False при определени условия. По-долу е посочен пример, когато има задачи, изпълнявани на управляващия сървър, но ще считаме тази заявка за неуспешна (ще зададем променливата success на False, въпреки че API повикването беше успешно и върна код 200).

for task in task_result.data["tasks"]:
    if task["status"] == "failed" or task["status"] == "partially succeeded":
        task_result.set_success_status(False)
        break

response()

Методът response позволява да видите речник с код на отговор (status_code) и с тяло на отговора (body).

In [94]: api_versions.response()                                                
Out[94]: 
{'status_code': 200,
 'data': {'current-version': '1.6',
  'supported-versions': ['1', '1.1', '1.2', '1.3', '1.4', '1.5', '1.6']}}

data

Позволява да видите само тялото на отговора (body) без излишна информация.

In [93]: api_versions.data                                                      
Out[93]: 
{'current-version': '1.6',
 'supported-versions': ['1', '1.1', '1.2', '1.3', '1.4', '1.5', '1.6']}

error_message

Тази информация е налична, само когато при обработката на API заявката е възникнала грешка (код на отговор не 200). Пример за изход

In [107]: api_versions.error_message                                            
Out[107]: 'code: generic_err_invalid_parameter_namenmessage: Unrecognized parameter [1]n'

Полезни примери

По-долу са изброени примери, при които се използват API повиквания, добавени във версия Management API 1.6.

Първо, нека разгледаме работата на повикванията add-host и add-address-range. Да приемем, че трябва да създадем като обекти от тип хост всички IP адреси в подсет 192.168.0.0/24, чийто последен октет е 5, а всички останали IP адреси да запишем като обекти от тип диапазон адреси. Като това, адресът на подсета и широковещателният адрес ще бъдат изключени.

И така, по-долу е представен скрипт, в който се решава този проблем и се създават 50 обекта от тип хост и 51 обект от тип диапазон адреси. За решаването на задачата са необходими 101 API повиквания (без финалното повикване publish). Също така с помощта на модула timeit изчисляваме времето за изпълнение на скрипта преди публикуването на промените.

Скрипт с използване на add-host и add-address-range

import timeit
from cpapi import APIClient, APIClientArgs

start = timeit.default_timer()

first_ip = 1
last_ip = 4

client_args = APIClientArgs(server="192.168.47.240")

with APIClient(client_args) as client: 
     login = client.login_with_api_key('3TsbPJ8ZKjaJGvFyoFqHFA==')
     for ip in range(5,255,5):
         add_host = client.api_call("add-host", {"name" : f"h_192.168.0.{ip}", "ip-address": f'192.168.0.{ip}'})
     while last_ip < 255:
         add_range = client.api_call("add-address-range", {"name": f"r_192.168.0.{first_ip}-{last_ip}", "ip-address-first": f"192.168.0.{first_ip}", "ip-address-last": f"192.168.0.{last_ip}"})
         first_ip+=5
         last_ip+=5
     stop = timeit.default_timer() 
     publish = client.api_call("publish")
     
print(f'Time to execute batch request: {stop - start} seconds')

В моя лабораторна среда, изпълнението на този скрипт отнема от 30 до 50 секунди в зависимост от натоварването на управляващия сървър.

А сега да разгледаме как да решим същата задача с помощта на API повикване add-objects-batch, чиято поддръжка е добавена в версия API 1.6. Тази команда позволява с една API заявка да създадете веднага много обекти. Освен това, те могат да бъдат обекти от различни типове (например хостове, подсети и диапазони адреси). Така нашата задача може да бъде решена в рамките на едно API повикване.

Скрипт с използване на add-objects-batch

import timeit
from cpapi import APIClient, APIClientArgs

start = timeit.default_timer()

client_args = APIClientArgs(server="192.168.47.240")

objects_list_ip = []
objects_list_range = []

for ip in range(5,255,5):
    data = {"name": f'h_192.168.0.{ip}', "ip-address": f'192.168.0.{ip}'}
    objects_list_ip.append(data)
    
first_ip = 1
last_ip = 4


while last_ip < 255:
    data = {"name": f"r_192.168.0.{first_ip}-{last_ip}", "ip-address-first": f"192.168.0.{first_ip}", "ip-address-last": f"192.168.0.{last_ip}"}
    objects_list_range.append(data)
    first_ip+=5
    last_ip+=5

data_for_batch = {
  "objects" : [ {
    "type" : "host",
    "list" : objects_list_ip
}, {
    "type" : "address-range",
    "list" : objects_list_range
  }]
}


with APIClient(client_args) as client: 
     login = client.login_with_api_key('3TsbPJ8ZKjaJGvFyoFqHFA==')
     add_objects_batch = client.api_call("add-objects-batch", data_for_batch)
     stop = timeit.default_timer() 
     publish = client.api_call("publish")
     
print(f'Time to execute batch request: {stop - start} seconds')

И изпълнението на този скрипт в моята лабораторна среда отнема от 3 до 7 секунди в зависимост от натоварването на управляващия сървър. Тоест, средно, при 101 обекта, API повикването от тип batch работи 10 пъти по-бързо. При по-висок брой обекти разликата ще бъде още по-впечатляваща.

Сега нека да разгледаме как да работим с set-objects-batch. С помощта на това API повикване можем масово да променим който и да е параметър. Нека настроим първата половина от адресите от предишния пример (до .124 на хоста, а също и на диапазоните) на цвят sienna, а на втората половина адресите да зададем цвят khaki.

Промяна на цвета на обектите, създадени в предишния пример

от cpapi импортирайте APIClient, APIClientArgs

client_args = APIClientArgs(server="192.168.47.240")

objects_list_ip_first = []
objects_list_range_first = []
objects_list_ip_second = []
objects_list_range_second = []

за ip в диапазона(5,125,5):
    data = {"name": f'h_192.168.0.{ip}', "color": "sienna"}
    objects_list_ip_first.append(data)
    
за ip в диапазона(125,255,5):
    data = {"name": f'h_192.168.0.{ip}', "color": "khaki"}
    objects_list_ip_second.append(data)
    
first_ip = 1
last_ip = 4
докато last_ip < 125:
    data = {"name": f"r_192.168.0.{first_ip}-{last_ip}", "color": "sienna"}
    objects_list_range_first.append(data)
    first_ip+=5
    last_ip+=5
    
докато last_ip < 255:
    data = {"name": f"r_192.168.0.{first_ip}-{last_ip}", "color": "khaki"}
    objects_list_range_second.append(data)
    first_ip+=5
    last_ip+=5

data_for_batch_first  = {
  "objects" : [ {
    "type" : "host",
    "list" : objects_list_ip_first
}, {
    "type" : "address-range",
    "list" : objects_list_range_first
  }]
}

data_for_batch_second  = {
  "objects" : [ {
    "type" : "host",
    "list" : objects_list_ip_second
}, {
    "type" : "address-range",
    "list" : objects_list_range_second
  }]
}

с APIClient(client_args) като клиент: 
     login = client.login_with_api_key('3TsbPJ8ZKjaJGvFyoFqHFA==') 
     set_objects_batch_first = client.api_call("set-objects-batch", data_for_batch_first)
     set_objects_batch_second = client.api_call("set-objects-batch", data_for_batch_second)
     publish = client.api_call("publish")

Можете да изтриете множество обекти в един API повикване с помощта на delete-objects-batch. А сега нека разгледаме примерен код, който премахва всички хостове, създадени по-рано чрез add-objects-batch.

Изтриване на обекти с помощта на delete-objects-batch

от cpapi импортирайте APIClient, APIClientArgs

client_args = APIClientArgs(server="192.168.47.240")

objects_list_ip = []
objects_list_range = []

за ip в диапазона(5,255,5):
    data = {"name": f'h_192.168.0.{ip}'}
    objects_list_ip.append(data)

first_ip = 1
last_ip = 4
докато last_ip < 255:
    data = {"name": f"r_192.168.0.{first_ip}-{last_ip}"}
    objects_list_range.append(data)
    first_ip+=5
    last_ip+=5

data_for_batch = {
  "objects" : [ {
    "type" : "host",
    "list" : objects_list_ip
}, {
    "type" : "address-range",
    "list" : objects_list_range
  }]
}

с APIClient(client_args) като клиент: 
     login = client.login_with_api_key('3TsbPJ8ZKjaJGvFyoFqHFA==')
     delete_objects_batch = client.api_call("delete-objects-batch", data_for_batch)
     publish = client.api_call("publish")

print(delete_objects_batch.data)

Всички функции, които се появяват в новите версии на софтуера Check Point, веднага получават и API повиквания. Така, в R80.40 се появиха функции като Revert to revision и Smart Task, за които веднага бяха подготвени съответните API повиквания. Освен това, цялата функционалност при преминаване от Legacy конзоли в режим Unified Policy също получава поддръжка на API. Например, дългоочакваното обновление в версията R80.40 беше преместването на политиката за HTTPS Inspection от Legacy режим в режим Unified Policy, и тази функционалност веднага получи API повиквания. Ето пример за код, който добавя на върха на политиката за HTTPS Inspection правило, което изключва от инспекция 3 категории (Здраве, Финанси, Държавни услуги), които е забранено да се инспектират съгласно законодателството в редица държави.

Добавяне на правило в политиката за HTTPS Inspection

from cpapi import APIClient, APIClientArgs

client_args = APIClientArgs(server="192.168.47.240")

data = {
  "layer" : "Default Layer",
  "position" : "top",
  "name" : "Legal Requirements",
  "action": "bypass",
  "site-category": ["Health", "Government / Military", "Financial Services"]
}

with APIClient(client_args) as client: 
     login = client.login_with_api_key('3TsbPJ8ZKjaJGvFyoFqHFA==')
     add_https_rule = client.api_call("add-https-rule", data)
     publish = client.api_call("publish")

Стартиране на Python скриптове на управленския сървър на Check Point

Всичко е същото README.md съдържа информация как да стартирате скриптове на Python директно от управленския сървър. Това може да бъде удобно, когато нямате възможност да се свържете с API сървъра от друга машина. Записах шестминутно видео, в което разглеждам инсталацията на модула cpapi и особеностите на стартиране на Python скриптове на управленския сървър. Като пример се стартира скрипт, който автоматизира конфигурацията на нов шлюз за задача като мрежов одит Security CheckUp. От особеностите, с които се сблъсках: в версия Python 2.7 все още не е налична функцията input, затова за обработка на информацията, която въвежда потребителят, се използва функцията raw_input. В останалото, кодът е същият като за стартиране от други машини, само че е по-удобно да се използва функцията login_as_root, за да не е необходимо да посочвате собствените си потребителско име, парола и IP адрес на управленския сървър отново.

Пуснете видеото

Скрипт за бърза настройка на Security CheckUp

от __future__ импортиране print_function
импорт getpass
импорт sys, os
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
от cpapi импортиране APIClient, APIClientArgs

def main():
    с APIClient() като client:
       # ако client.check_fingerprint() е False:
       #     print("Не успяхме да получим пръстовия отпечатък на сървъра - Проверете свързаността със сървъра.")
       #     exit(1)
        login_res = client.login_as_root()

        ако login_res.success е False:
            print("Неуспешно влизане:n{}".форматирай(login_res.error_message))
            exit(1)

        gw_name = raw_input("Въведете името на шлюза:")
        gw_ip = raw_input("Въведете IP адреса на шлюза:")
        ако sys.stdin.isatty():
            sic = getpass.getpass("Въведете еднократната парола за шлюза(SIC): ")
        иначе:
            print("Внимание! Вашата парола ще се покаже на екрана!")
            sic = raw_input("Въведете еднократната парола за шлюза(SIC): ")
        версия = raw_input("Въведете версията на шлюза(като RXX.YY):")
        add_gw = client.api_call("add-simple-gateway", {'name' : gw_name, 'ipv4-address' : gw_ip, 'one-time-password' : sic, 'version': version.capitalize(), 'application-control' : 'true', 'url-filtering' : 'true', 'ips' : 'true', 'anti-bot' : 'true', 'anti-virus' : 'true', 'threat-emulation' : 'true'})
        ако add_gw.success и add_gw.data['sic-state'] != "communicating":
            print("Сигурната връзка с шлюза не е установена!")
            exit(1)
        иначе, ако add_gw.success:
            print("Шлюзът беше добавен успешно.")
            gw_uid = add_gw.data['uid']
            gw_name = add_gw.data['name']
        иначе:
            print("Неуспех при добавяне на шлюза - {}".форматирай(add_gw.error_message))
            exit(1)

        change_policy = client.api_call("set-access-layer", {"name" : "Network", "applications-and-url-filtering": "true", "content-awareness": "true"})
        ако change_policy.success:
            print("Политиката беше променена успешно")
        иначе:
            print("Неуспех при промяна на политиката - {}".форматирай(change_policy.error_message))
        change_rule = client.api_call("set-access-rule", {"name" : "Правило за почистване", "layer" : "Network", "action": "Accept", "track": {"type": "Detailed Log", "accounting": "true"}})
        ако change_rule.success:
            print("Правилото за почистване беше променено успешно")
        иначе:
            print("Неуспех при промяна на правилото за почистване - {}".форматирай(change_rule.error_message))

        # публикуване на резултата
        publish_res = client.api_call("publish", {})
        ако publish_res.success:
            print("Промените бяха публикувани успешно.")
        иначе:
                print("Неуспех при публикуване на промените - {}".форматирай(install_tp_policy.error_message))

        install_access_policy = client.api_call("install-policy", {"policy-package" : "Standard", "access" : 'true',  "threat-prevention" : 'false', "targets" : gw_uid})
        ако install_access_policy.success:
            print("Достъпната политика беше инсталирана")
        иначе:
                print("Неуспех при инсталиране на достъпната политика - {}".форматирай(install_tp_policy.error_message))

        install_tp_policy = client.api_call("install-policy", {"policy-package" : "Standard", "access" : 'false',  "threat-prevention" : 'true', "targets" : gw_uid})
        ако install_tp_policy.success:
            print("Политиката за предотвратяване на заплахи беше инсталирана")
        иначе:
            print("Неуспех при инсталиране на политиката за предотвратяване на заплахи - {}".форматирай(install_tp_policy.error_message))
        
        # добавяне на пароли и фрази в речника
        с отворен('additional_pass.conf') като f:
            line_num = 0
            за ред в f:
                line_num += 1
                add_password_dictionary = client.api_call("run-script", {"script-name" : "Добави пароли и фрази", "script" : "printf "{}" >> $FWDIR/conf/additional_pass.conf".форматирай(line), "targets" : gw_name})
                ако add_password_dictionary.success:
                    print("Редът на речника с пароли {} беше добавен успешно".форматирай(line_num))
                иначе:
                    print("Неуспех при добавяне на речника - {}".форматирай(add_password_dictionary.error_message))

main()

Примерен файл с речник на паролите additional_pass.conf
{
"passwords" : ["вредоносен софтуер","зловреден","инфектиран","Инфектиран"],
"phrases" : ["парола","Парола","Парола","пар","код","ключ","pwd","password","ключ","ключ","шифър","Шифър"]
}

Заключение

Тази статия разглежда само основните възможности за работа Python SDK и модула cpapi(както може да се досетите, това всъщност са синоними), и изучавайки кода в този модул, вие ще откриете още повече възможности за работа с него. Не е изключено да почувствате желание да го допълвате с ваши класове, функции, методи и променливи. Винаги можете да споделяте вашите разработки и да преглеждате други скриптове за Check Point в секцията CodeHub в общността CheckMates, която обединява както разработчиците на продукти, така и потребителите.

Приятно кодене и благодаря, че прочетохте до края!

Източник: habr.com

Купете надежден хостинг за сайтове с защита от DDoS, VPS VDS сървъри 🔥 Купете надежден хостинг за сайтове с защита от DDoS, VPS VDS сървъри | ProHoster