Python Coding : Call Salesforce API using OAuth 2.0 JWT Bearer token Flow 以 Python 代碼實現Salesforce JWT 伺服器簽名驗證

Prequisites : Setup Connected App with  X509 Certification.

預設需要先在Salesforce 上載X509證書。詳細設定連接。（可接受自行簽署版本證書)

Details Set up in Salesforce side : link

#POC of JWT call to API#Colab install library with >> !pip install pyJWT[crypto]

import jwt
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
import time
import requests
import json

expTime = int(time.time()) + 900  #Unix Epoch Timestamp Expire time. 15mins buffer
jwt_algorithm = 'RS256'
jwt_private_key_file = '/myPath/server.key'
jwt_payload = {
  "iss": "clientid",
  "sub": "salesforce_username_email",
  "aud": "https://test.salesforce.com",
  "exp": expTime
}

jwt_header = {
  "alg": "RS256",
  "typ": "JWT"
}
salesforceEndpoint = 'https://test.salesforce.com/services/oauth2/token'
api_endpoint = 'https://instance_domain.my.salesforce.com/services/data/v57.0/sobjects/myObject__c'

test_data={
    "description": "PythonSent",
    "language": "en",
    "remarks": "xxx",
    "expiry_time": expTime+900,
    "created_time": expTime
}


def loginSfdc():
    # Load the private key 
    with open(jwt_private_key_file, 'r') as f:
        jwt_private_key = f.read()

    # Generate the JWT token
    jwt_token = jwt.encode(jwt_payload, jwt_private_key, algorithm=jwt_algorithm,headers=jwt_header)
    print('token : ' + jwt_token);
    salesforceUrlParams = {
        'grant_type' : 'urn:ietf:params:oauth:grant-type:jwt-bearer',
        'assertion' :jwt_token
    }

    response = requests.post(salesforceEndpoint, params=salesforceUrlParams, verify=False)

    if response.status_code == 200:
        authToken = response.json()['access_token']
        print('return token ' + authToken)
        return authToken
    else:
        print('Request failed with status code:', response.status_code)
        print('Request failed with status code:', response.text)
        return(0)

#function to call
def callSfdc_pocApi(authToken):
    session = requests.Session()
    rheaders = {
    'Authorization': 'Bearer ' + authToken,
    'Content-Type': 'application/json'
    }
   
    response = requests.post(api_endpoint,headers=rheaders, data=json.dumps(test_data))
    returnStr = ''
    if response.status_code == 200:
        data = response.json()
        returnStr = data
    else:
        print('Request failed with status code:', response.text)
        returnStr = response.status_code

    return returnStr

#Main
response1 = loginSfdc()
if not response1 == 0:
    response2 = callSfdc_pocApi(response1)
    print ("OK")
    print(response2)
else:
    print ('Failed to Login')