Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
__pycache__/
.venv/
34 changes: 16 additions & 18 deletions docs/scripts/api-usage-example-python/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@
import traceback
import sched
import time
import base64
import mgmt_api
scheduler = sched.scheduler(time.time, time.sleep)
from mgmt_api_helper_functions import *
Expand Down Expand Up @@ -43,33 +42,21 @@ def main():
})
print("Creating Web API Asset and components\n")

modes = [
main_mode = "Prevent"
sub_practice_modes = [
{
"mode": "AccordingToPractice",
"subPractice": "IPS"
},
{
"mode": "Prevent"
},
{
"mode": "AccordingToPractice",
"subPractice": "SchemaValidation"
},
{
"mode": "AccordingToPractice",
"subPractice": "APIAttacks"
}
]
practice_name = "{} {}".format("ACME Web API Practice", generate_full_random_hex())
file = open("D:\git\\test-mgmt-versions-api\OpenAPISchema.yml", "r")
file_content = file.read()
message_bytes = file_content.encode('ascii')
base64_bytes = base64.b64encode(message_bytes)
header64 = "OpenAPISchema.yml;$$:$$;data:application/octet-stream;base64," + base64_bytes.decode('ascii')
web_api_practice_id = mgmt_api.create_web_api_practice(modes, {
web_api_practice_id = mgmt_api.create_web_api_practice(main_mode, sub_practice_modes, {
"name": practice_name,
"visibility": "Shared",
"SchemaValidation": {"OasSchema": header64}
"visibility": "Shared"
})

web_api_practice_obj = {
Expand Down Expand Up @@ -98,8 +85,19 @@ def main():

print("Creating Web Application Asset and components\n")

web_app_main_mode = "Prevent"
web_app_sub_practice_modes = [
{
"mode": "AccordingToPractice",
"subPractice": "IPS"
},
{
"mode": "AccordingToPractice",
"subPractice": "WebAttacks"
}
]
web_application_practice_name = "{} {}".format(PRACTICE_BASE_NAME, generate_full_random_hex())
web_application_practice_id = mgmt_api.create_web_application_practice(modes, {
web_application_practice_id = mgmt_api.create_web_application_practice(web_app_main_mode, web_app_sub_practice_modes, {
"name": web_application_practice_name,
"visibility": "Shared"
})
Expand Down
19 changes: 13 additions & 6 deletions docs/scripts/api-usage-example-python/mgmt_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -283,36 +283,43 @@ def delete_behavior(behavior_id):
print("Deleting Behavior: ", behavior_id, res, "\n")


def create_web_application_practice(modes, practice_input):
def create_web_application_practice(main_mode, sub_practice_modes, practice_input):
"""
Executes a "create web application practice" mutation with graphql-client
:param asset_id: The id of the asset to associate the practice with
:param main_mode: main practice mode (e.g., "Prevent")
:param sub_practice_modes: sub practices modes
:param practice_input: the practice input
:return: The new practice id
"""
practice_variables = {
"modes": modes,
"mainMode": main_mode,
"subPracticeModes": sub_practice_modes,
"practiceInput": practice_input
}
practice_res = perform_infinity_request(query_or_mutation=CREATE_WEB_APPLICATION_PRACTICE_MUTATION,
version=1,
variables=practice_variables,
description="Creating Practice")
practice_id = practice_res['data'][CREATE_WEB_APPLICATION_PRACTICE_MUTATION_NAME]['id']
print("Creating Web Application Practice")
return practice_id


def create_web_api_practice(modes, practice_input):
def create_web_api_practice(main_mode, sub_practice_modes, practice_input):
"""
Executes a "create web api practice" mutation with graphql-client
:param modes: sub practices modes
:param main_mode: main practice mode (e.g., "Prevent")
:param sub_practice_modes: sub practices modes
:param practice_input: the practice input
:return: The new practice id
"""
practice_variables = {
"modes": modes,
"mainMode": main_mode,
"subPracticeModes": sub_practice_modes,
"practiceInput": practice_input
}
practice_res = perform_infinity_request(query_or_mutation=CREATE_WEB_API_PRACTICE_MUTATION,
version=1,
variables=practice_variables,
description="Creating Practice")
practice_id = practice_res['data'][CREATE_WEB_API_PRACTICE_MUTATION_NAME]['id']
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -224,7 +224,7 @@ def create_session(response_type, description):
retry_strategy = CallbackRetry(
total=vars.REQUESTS_NUMBER_OF_RETRIES,
status_forcelist=vars.REQUESTS_STATUS_CODES_RETRY,
method_whitelist=vars.REQUESTS_METHODS_LIST_RETRY,
allowed_methods=vars.REQUESTS_METHODS_LIST_RETRY,
backoff_factor=vars.REQUESTS_EXP_SLEEP_RETRY,
raise_on_status=False,
callback=add_retry_response_and_status_code,
Expand Down Expand Up @@ -328,7 +328,7 @@ def execute(self, query: str, variables: dict = None,
request_body = self.__request_body(
query=query, variables=variables, operation_name=operation_name
)

print("query : {0}, variables: {1}".format(query, variables))
result = requests.post(
self.endpoint,
json=request_body,
Expand All @@ -346,7 +346,7 @@ def create_session(response_type, description):
retry_strategy = CallbackRetry(
total=vars.REQUESTS_NUMBER_OF_RETRIES,
status_forcelist=vars.REQUESTS_STATUS_CODES_RETRY,
method_whitelist=vars.REQUESTS_METHODS_LIST_RETRY,
allowed_methods=vars.REQUESTS_METHODS_LIST_RETRY,
backoff_factor=vars.REQUESTS_EXP_SLEEP_RETRY,
raise_on_status=False,
callback=add_retry_response_and_status_code,
Expand Down
39 changes: 22 additions & 17 deletions docs/scripts/api-usage-example-python/vars.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,33 +11,41 @@
# prod
CLIENT_ID = ""
SECRET_KEY = ""
region = "US"# US or EU
region = "US" # US or EU

CLOUD_INFRA_HOST = "https://cloudinfra-gw-us.portal.checkpoint.com" if region == 'US' else "https://cloudinfra-gw.portal.checkpoint.com"
APP_PATH = "/app/i2"
CLOUD_INFRA_HOST = (
"https://cloudinfra-gw-us.portal.checkpoint.com"
if region == "US"
else "https://cloudinfra-gw.portal.checkpoint.com"
)
APP_PATH = "/app/waf"

AUTH_PATH = "/auth/external"



REQUESTS_NUMBER_OF_RETRIES = 0
REQUESTS_STATUS_CODES_RETRY = [429, 500, 502, 503, 504, 400, 401, 403, 404]
REQUESTS_METHODS_LIST_RETRY = ['POST', 'DELETE', 'GET', 'PUT', 'OPTIONS', 'HEAD', 'TRACE']
REQUESTS_METHODS_LIST_RETRY = [
"POST",
"DELETE",
"GET",
"PUT",
"OPTIONS",
"HEAD",
"TRACE",
]
REQUESTS_EXP_SLEEP_RETRY = 2
SCHEDULER_WAIT_BEFORE_FIRST_RUN_SECONDS = 10

# content type
CONTENT_TYPE = "application/json"

auth_request_body = {
"clientId": CLIENT_ID,
"accessKey": SECRET_KEY
}
auth_request_body = {"clientId": CLIENT_ID, "accessKey": SECRET_KEY}

all_requests_headers = {
"user-agent": "Infinity Next Sanity Script",
"Content-Type": CONTENT_TYPE,
"Authorization": ""
"Authorization": "",
}

GRAPHQL_CLIENT_ENDPOINT = "{}{}/graphql".format(CLOUD_INFRA_HOST, APP_PATH)
Expand Down Expand Up @@ -263,15 +271,15 @@

# ----------PRACTICES mutations----------

CREATE_WEB_APPLICATION_PRACTICE_MUTATION = """mutation newWebApplicationPractice($ownerId: ID, $modes: [PracticeModeInput], $practiceInput: WebApplicationPracticeInput){
newWebApplicationPractice(ownerId: $ownerId, modes: $modes, practiceInput: $practiceInput){
CREATE_WEB_APPLICATION_PRACTICE_MUTATION = """mutation newWebApplicationPractice($ownerId: ID, $mainMode: PracticeMode, $subPracticeModes: [PracticeModeInput], $practiceInput: WebApplicationPracticeInput){
newWebApplicationPractice(ownerId: $ownerId, mainMode: $mainMode, subPracticeModes: $subPracticeModes, practiceInput: $practiceInput){
id
name
}
}"""

CREATE_WEB_API_PRACTICE_MUTATION = """mutation newWebAPIPractice($ownerId: ID, $modes: [PracticeModeInput], $practiceInput: WebAPIPracticeInput){
newWebAPIPractice(ownerId: $ownerId, modes: $modes, practiceInput: $practiceInput){
CREATE_WEB_API_PRACTICE_MUTATION = """mutation newWebAPIPractice($ownerId: ID, $mainMode: PracticeMode, $subPracticeModes: [PracticeModeInput], $practiceInput: WebAPIPracticeInput){
newWebAPIPractice(ownerId: $ownerId, mainMode: $mainMode, subPracticeModes: $subPracticeModes, practiceInput: $practiceInput){
id
name
}
Expand Down Expand Up @@ -345,9 +353,6 @@
SLEEP_AFTER_CHECK_ENFORCE_STATUS = 10





def clean_last_token_from_headers():
global all_requests_headers
all_requests_headers["Authorization"] = ""
Loading