API Access with KONG

import subprocess
import argparse
import json
import base64
import time
import requests
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
from jose import jwt
 
def convert_pkcs1_to_pkcs8(input_key_path):
# Execute the openssl command to convert PKCS#1 to PKCS#8 and
capture the output
command = [
'openssl', 'pkcs8', '-topk8', '-inform', 'PEM', '-outform', 'PEM',
'-in', input_key_path, '-nocrypt'
]
 
def load_private_key(pkcs8_key_data):
private_key = serialization.load_pem_private_key(
pkcs8_key_data.encode(),
password=None,
backend=default_backend()
)
return private_key
 
def convert_to_jwk(private_key):
numbers = private_key.private_numbers()
jwk_dict = {
"kty": "RSA",
"n":
base64.urlsafe_b64encode(numbers.public_numbers.n.to_bytes((numbers.public_numbers.n.bit_length() + 7) // 8, byteorder='big')).decode('utf-8').rstrip("="),
"e":
base64.urlsafe_b64encode(numbers.public_numbers.e.to_bytes((numbers.public_numbers.e.bit_length() + 7) // 8, byteorder='big')).decode('utf-8').rstrip("="),
"d":
base64.urlsafe_b64encode(numbers.d.to_bytes((numbers.d.bit_length() + 7) // 8, byteorder='big')).decode('utf-8').rstrip("="),
"p":
base64.urlsafe_b64encode(numbers.p.to_bytes((numbers.p.bit_length() + 7) // 8, byteorder='big')).decode('utf-8').rstrip("="),
"q":
base64.urlsafe_b64encode(numbers.q.to_bytes((numbers.q.bit_length() + 7) // 8, byteorder='big')).decode('utf-8').rstrip("="),
"dp":
base64.urlsafe_b64encode(numbers.dmp1.to_bytes((numbers.dmp1.bit_length() + 7) // 8, byteorder='big')).decode('utf-8').rstrip("="),
"dq":
base64.urlsafe_b64encode(numbers.dmq1.to_bytes((numbers.dmq1.bit_length() + 7) // 8, byteorder='big')).decode('utf-8').rstrip("="),
"qi":
base64.urlsafe_b64encode(numbers.iqmp.to_bytes((numbers.iqmp.bit_length() + 7) // 8, byteorder='big')).decode('utf-8').rstrip("=")
numbers = private_key.private_numbers()
jwk_dict = {
 
def generate_payload(env, app_id):
 
if env == 'stage':
aud = 'https://sso.qa.prudential.com:443/am/oauth2'
elif env == 'prod':
aud = 'https://sso.prudential.com:443/am/oauth2'
else:
raise ValueError(f"Invalid environment: {env}, valid envs are
qa, stage or prod.")
 
exp = int(time.time()) + 300
jti = subprocess.run(['openssl', 'rand', '-base64', '16'],
capture_output=True, text=True).stdout.strip()
 
payload = {
"iss": app_id,
"sub": app_id,
"aud": aud,
"exp": exp,
"jti": jti
}
 
payload_json = json.dumps(payload, separators=(',', ':'))
 
# Save payload to a file
with open('payload.json', 'w') as f:
f.write(payload_json)
 
return payload_json
 
def create_assertion(payload, private_key):
headers = {
'alg': 'RS256'
}
token = jwt.encode(json.loads(payload), private_key, algorithm='RS256', headers=headers)
return token
 
def get_access_token(env, assertion, scope):
if env == 'stage':
url = 'https://api-stage.prudential.com/gt/v1/iam/oauth/token'
elif env == 'prod':
url = 'https://api.prudential.com/gt/v1/iam/oauth/token'
else:
raise ValueError(f"Invalid environment: {env}, valid envs are qa, stage or prod.")
headers = {
'Content-Type': 'application/x-www-form-urlencoded'
}
data = {
'grant_type': 'client_credentials',
'client_assertion': assertion,
'client_assertion_type': 'urn:ietf:params:oauth:client-assertion-type:jwt-bearer',
'scope': scope
}
response = requests.post(url, headers=headers, data=data)
response.raise_for_status() # Raise an exception for HTTP errors
return response.json()
 
def main():
parser = argparse.ArgumentParser(description="Convert PKCS#1
private key to JWK, generate payload, create JWT assertion, and get access token.")
parser.add_argument('env', type=str, help='Environment')
parser.add_argument('input_key_path', type=str, help='Path to the
PKCS#1 private key file')
parser.add_argument('app_client_id', type=str, help='Application
Client ID')
parser.add_argument('scope', type=str, help='Scope for the token')
args = parser.parse_args()
 
env = args.env
input_key_path = args.input_key_path
app_client_id = args.app_client_id
scope = args.scope
 
# Step 1: Convert PKCS#1 to PKCS#8 and get the result as a string
pkcs8_key_data = convert_pkcs1_to_pkcs8(input_key_path)
 
# Step 2: Load the PKCS#8 key from the string data
private_key = load_private_key(pkcs8_key_data)
 
# Step 3: Convert to JWK
jwk_dict = convert_to_jwk(private_key)
 
# Convert the JWK dictionary to a JSON string
jwk_json = json.dumps(jwk_dict, separators=(',', ':'))
 
# Step 4: Generate the authentication payload
payload_json = generate_payload(env, app_client_id)
 
# Step 5: Create the JWT assertion
assertion = create_assertion(payload_json, pkcs8_key_data)
 
# Print the assertion
print(json.dumps(token_response, indent=4))
 
if __name__ == "__main__":
main()