Notes

import boto3
from datetime import datetime, timedelta
import json
import os
import urllib3
from urllib3.exceptions import HTTPError
from urllib.parse import urlencode
from urllib3.util.retry import Retry
from urllib3.exceptions import MaxRetryError


retry_strategy = Retry(
total=5, # Total number of retries
backoff_factor=0.3, # Wait time between retries (exponential backoff)
status_forcelist=[502, 503, 504], # Status codes to retry on
allowed_methods=["HEAD", "GET", "OPTIONS", "POST", "PUT", "DELETE", "TRACE"] # Methods to retry
)

def format_date(date):
return date.strftime('%Y-%m-%d %H:%M:%S')

def get_secret(secret_name, region):
client = boto3.client('secretsmanager', region_name=region)
response = client.get_secret_value(SecretId=secret_name)
secret = json.loads(response['SecretString'])
token = get_token(secret)
if 'Error ' in token:
return 'Invalid Token'
return token

def get_token(secret):
http = urllib3.PoolManager()
url = os.getenv('cs_sn_oauth_url')
headers = {
'Content-Type': 'application/x-www-form-urlencoded'
}
data = {
'grant_type': 'refresh_token',
'client_id': secret['client_id'],
'client_secret': secret['client_secret'],
'refresh_token': secret['refresh_token']
}
json_data = urlencode(data)

try:
response = http.request('POST',url, body=json_data, headers=headers)
if response.status == 200:
access_token = json.loads(response.data.decode('utf-8'))['access_token'].strip()
print("Access Token:", access_token)
return access_token
else:
print(f"Error: Received status code {response.status}")
err = response.data.decode('utf-8')
print(err)
except Exception as err:
print("TOKEN ERROR")
print(err)
return 'Error :' + err

def get_parameter(name, region):
client = boto3.client('ssm', region_name=region)
response = client.get_parameter(Name=name, WithDecryption=True)
return json.loads(response['Parameter']['Value'])

def createServiceNowStandardChange():
url = os.getenv("cg_sn_url")
ca_certs='/opt/python/certs/cg_ca_certs.crt'
if not os.path.isfile(ca_certs):
print("Ca certificates are not found")
return None
secretName = os.getenv("sn_oauth_secrets")
http = urllib3.PoolManager(
cert_reqs='CERT_REQUIRED',
ca_certs=ca_certs
)
region = 'us-east-1';
now = datetime.now()
date_start = format_date(now)
date_end = format_date(now + timedelta(days=1)) # 1 day from now

parameters = get_parameter('/onyx/servicenow/parameters', region)
template_name = parameters['template_name']
requested_by = parameters['requested_by']
assigned_to = parameters['assigned_to']
business_service = parameters['business_service']
short_description = parameters['short_description']

data = {
"template_name": template_name,
"fields": {
"requested_by": requested_by,
"assigned_to": assigned_to,
"u_conflict_justification": "No impact on critical services",
"business_service": business_service,
"description": "Allow new service to publish data",
"short_description": short_description,
"cmdb_ci": "Banzai",
"start_date": date_start,
"end_date": date_end,
}
}
token = get_secret(secretName, region)
if token == 'Invalid Token':
return None
headers = {
'Content-Type': 'application/json',
'Cache-Control' : 'no-cache',
'Authorization': f'Bearer {token}'
}
json_data = json.dumps(data)

try:
print("Calling ServiceNow")
response = http.request('POST', url, body=json_data, headers=headers)
print("SERVICENOW SUCCESS")
print(response.status)
if response.status == 400:
print("Bad Request: Check your request parameters.")
result = response.data.decode('utf-8')
dict_data = json.loads(result)
return dict_data
except MaxRetryError as e:
print("Max retries exceeded")
print(e)
except HTTPError as e:
print(f"HTTP error occurred: {e}")
except Exception as err :
print("SERVICENOW ERROR",err)
return None