附註：

使用 Microsoft Entra ID 的存取權杖安全地存取 Oracle Integration

簡介

隨著 Oracle Integration 客戶採用多雲端策略，他們通常需要連結不同雲端提供者的業務應用程式和流程。例如，公司可能有一個在 Microsoft Azure 上執行的應用程式，需要從 Oracle Cloud Infrastructure 應用程式存取資料。一般而言，您會從 Oracle Cloud Infrastructure Identity and Access Management (OCI IAM) 取得權杖以擷取此資料。不過，使用多個雲端提供者表示處理多個權杖，這可能是複雜且造成安全風險。

想像一下，如果您可以使用一個 OAuth 權杖來與跨不同雲端提供者的應用程式整合，那就有多方便。本教學課程主要介紹如何使用第三方 OAuth 提供者來呼叫 Oracle Integration 流程。

架構

讓我們將解決方案流程視覺化：

影像 1

處理會從使用者或商業應用程式開始，從 Microsoft Entra ID 取得 OAuth 權杖。取得之後，此權杖會用來呼叫透過 OCI API Gateway 公開的端點。設定為使用自訂授權者 OCI 函數 (之前稱為 Oracle Functions) 的 OCI API 閘道會先呼叫此授權者函數來驗證權杖。驗證成功後，它會呼叫實際的後端端點 (即 Oracle Integration 流程)。

現在，讓我們深入了解實作此程序的詳細資料。為求簡化，我們將分為三個步驟：

為何要分別使用資源擁有者密碼證明資料 (ROPC) 和 JSON Web 權杖 (JWT) 宣告授權類型，才能從 Microsoft Entra ID 和 OCI IAM 取得存取權杖？

結合使用 ROPC 和 JWT 宣告授權，提供簡化且安全的方法來處理多雲端環境的認證和權杖交換。

適用對象

目標

必要條件

工作 1：向 Microsoft Entra ID 註冊應用程式

若要使用 Microsoft Entra ID 的 IAM 功能 (包括存取受保護資源 (圖表 API)，您必須註冊應用程式。

  1. 註冊應用程式。如需詳細資訊，請參閱在 Microsoft 識別平台註冊應用程式。

  2. 將總覽段落中的 Application (client) ID 值記下來。

    影像 2

  3. 請前往管理、憑證與加密密碼，然後新增從屬端加密密碼。將加密密碼值記下來，因為它將用於後續的工作中。

    影像 3

作業 2：OCI 識別網域中 JWT 使用者宣告的先決條件步驟

  1. 從此處完成「先決條件」工作： JWT 使用者宣告的先決條件。

  2. 驗證必要範圍的 Oracle Integration 應用程式之後，就會產生自行簽署金鑰組並設定機密應用程式。請記下 scope 值、private_key.pem 、 Client ID 和 Client Secret 。

    注意：將私密金鑰匯入為機密應用程式中的信任夥伴時，會使用與建立自行簽署金鑰組時所使用的相同 alias，並記下 alias 以供後續作業使用。

  3. 建立動態群組，以允許特定區間的資源類型 function 能夠從 OCI 保存庫服務讀取加密密碼。

    影像 6

作業 3：在 OCI 保存庫中建立加密密碼

使用 OCI 保存庫手動產生加密密碼選項儲存從作業 1 和作業 2 收集的加密密碼。如需詳細資訊，請參閱在保存庫中建立加密密碼。

影像 4

建立加密密碼之後，請從加密密碼資訊區段複製 OCID 值，然後將它儲存供後續作業使用。

影像 5

工作 4：建立和設定 func.py 檔案

我們將使用 OCI 函數作為自訂授權者來驗證 Microsoft Entra ID 存取權杖，並以 back_end_token 身分產生 OCI IAM 存取權杖。

  1. 若要啟動，請建立一個應用程式。在 OCI 函數中，應用程式是函數的邏輯群組。您為應用程式指定的特性會決定該應用程式中所有功能的資源配置與組態。如需詳細資訊，請參閱建立應用程式。

  2. 建立應用程式之後，將組態新增至應用程式。我們將從功能代碼中提取以下項目，使它更可攜且可配置，無需修改代碼。輸入索引鍵和值，然後按一下 + 。

    新增 Microsoft Entra ID、OCI 識別網域、作業 3 中收集之加密密碼的 OCID、別名、從作業 2 收集的範圍，以及驗證 microsoft Entra ID 權杖的圖表端點 https://graph.microsoft.com/v1.0/me。

    影像 7

  3. 若要建立函數，請前往入門，然後按一下啟動 OCI Cloud Shell ，在您的瀏覽器中開啟互動式 Linux 樣式 Cloud Shell。載入 OCI Cloud Shell 之後，您便可以直接從 OCI Cloud Shell 建立、開發及部署自訂授權端 Oracle 函數。

  4. 若要使用 Fn 專案命令行介面 (CLI) 建立函數，請為 Python 函數 fn init --runtime python MyCustomAuthorizer 輸入下列命令，然後按一下 Enter 。

    影像 8

  5. 已建立函數圖文模版，現在可據此編輯以包含自訂授權者邏輯。將目錄變更為函數資料夾，然後編輯 func.py 檔案。複製並貼上下列程式碼片段。

    影像 9

    import io
import json
import logging
import jwt
import datetime
from datetime import timedelta
import time
import base64

from fdk import response
import requests
from requests.auth import HTTPBasicAuth
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
import ociVault

oauth_apps = {}

def initContext(context):
    # This method takes elements from the Application Context and from OCI Vault to create the OAuth App Clients object.
    if (len(oauth_apps) < 2):
        try:
            logging.getLogger().info("initContext: Initializing context")

            oauth_apps['idcs'] = {'introspection_endpoint': context['idcs_token_endpoint'],
                                'client_id': context['idcs_app_client_id'],
                                'scope':context['idcs_oauth_scope'],
                                'alias':context['alias'],
                                'client_secret': ociVault.getSecret(context['idcs_client_secret_ocid'])}
            oauth_apps['AD'] = {'token_endpoint': context['ad_endpoint'],
                                'client_id': context['ad_app_client_id'],
                                'client_secret': ociVault.getSecret(context['ad_client_secret_ocid'])}

        except Exception as ex:
            logging.getLogger().error("initContext: Failed to get config or secrets" + str(ex))
            raise


def getAuthContext(token, client_apps):
    # This method populates the Auth Context that will be returned to the gateway.
    auth_context = {}
    access_token = token[len('Bearer '):]
    jwtToken = json.loads(json.dumps(jwt.decode(access_token, options={"verify_signature": False})))
    # Calling MSFT to validate the token
    try:
    logging.getLogger().info("getAuthContext: Calling Token Introspection function") 
    respIntrospectToken = introspectToken(access_token, client_apps['AD']['token_endpoint'], client_apps['AD']['client_id'], client_apps['AD']['client_secret'])
    except Exception as ex:
            logging.getLogger().error("getAuthContext: Failed to introspect token" + str(ex))
            raise

    # If AD confirmed the token valid and active, we can proceed to populate the auth context
    if (respIntrospectToken.status_code == 200):
        auth_context['active'] = True
        auth_context['principal'] = jwtToken['upn']
        auth_context['scope'] = 'https://graph.microsoft.com/.default'
        # Retrieving the back-end Token
        backend_token = getBackEndAuthToken(client_apps['idcs']['introspection_endpoint'], client_apps['idcs']['client_id'], client_apps['idcs']['client_secret'],client_apps['idcs']['scope'],client_apps['idcs']['alias'],auth_context['principal'])

        # The maximum TTL for this auth is the lesser of the API Client Auth (Entra ID) and the Gateway Client Auth (OCI IAM)
        if (datetime.datetime.fromtimestamp(jwtToken['exp']) < (datetime.datetime.utcnow() + timedelta(seconds=backend_token['expires_in']))):
            auth_context['expiresAt'] = (datetime.datetime.fromtimestamp(jwtToken['exp'])).replace(tzinfo=datetime.timezone.utc).astimezone().replace(microsecond=0).isoformat()
        else:
            auth_context['expiresAt'] = (datetime.datetime.utcnow() + timedelta(seconds=backend_token['expires_in'])).replace(tzinfo=datetime.timezone.utc).astimezone().replace(microsecond=0).isoformat()
        # Storing the back_end_token in the context of the auth decision so we can map it to Authorization header using the request/response transformation policy
        auth_context['context'] = {'back_end_token': ('Bearer ' + str(backend_token['access_token']))}

    else:
        # API Client token is not active, so we will go ahead and respond with the wwwAuthenticate header
        auth_context['active'] = False
        auth_context['wwwAuthenticate'] = 'Bearer realm=\"identity.oraclecloud.com\"'

    return(auth_context)

def introspectToken(access_token, introspection_endpoint, client_id, client_secret):
    # This method simply invokes the introspection api as configured in the configuration screen.  
    # The real validation happens in the getAuthContext function.  
    #payload = {'token': access_token}
    headers = {'Accept': 'application/json',
            'Authorization':'Bearer '+access_token}
    try:
        logging.getLogger().info("introspectToken: Introspecting Token") 
        resp = requests.get(introspection_endpoint,
                            headers=headers)
        print(resp)

    except Exception as ex:
        logging.getLogger().error("introspectToken: Failed to introspect token" + str(ex))
        raise

    return resp

def getBackEndAuthToken(token_endpoint, client_id, client_secret, scope, alias, principal):
    # This method gets the token from the back-end system (ORDS in this case)
    try:
        logging.getLogger().info("getBackEndAuthToken: Getting Backend Token") 
        print("Sub is " + principal)

        with open("private_key.pem", "rb") as key_file:
            private_key = serialization.load_pem_private_key(
                key_file.read(),
                password=None,
                backend=default_backend()
            )

        headers = {
            "alg": "RS256",
            "typ": "JWT",
            "kid": "abc"
        }

        claims = {
            "sub": principal,
            "aud": "https://identity.oraclecloud.com/",
            "iss": client_id,
            "iat": int(time.time()),
            "exp": int(time.time()) + 3600,  # 1 hour expiration
            "jti": "8c7df446-bfae-40be-be09-0ab55c655436"  # random number
        }

        logging.getLogger().info("Claims : ")
        logging.getLogger().info(claims) 

        jwt_assertion = jwt.encode(
            payload=claims,
            key=private_key,
            algorithm="RS256",
            headers=headers
        )
        logging.getLogger().info("Assertion is :") 
        logging.getLogger().info(jwt_assertion) 

        encoded = client_id + ":" + client_secret
        baseencoded = base64.urlsafe_b64encode(encoded.encode('UTF-8')).decode('ascii')
        payload = {'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
                'scope':scope, 'assertion':jwt_assertion}
        headers = {'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8', 'Authorization': 'Basic %s' % baseencoded, 'Accept': '*/*'}
        backend_token = json.loads(requests.post(token_endpoint,
                                                data=payload,
                                                headers=headers).text)
        logging.getLogger().info("Backend token in generated :") 
        logging.getLogger().info(backend_token) 

    except Exception as ex:
        logging.getLogger().error("getBackEndAuthToken: Failed to get ORDS token" + str(ex))
        raise

    return backend_token



def handler(ctx, data: io.BytesIO = None):
    initContext(dict(ctx.Config()))
    logging.getLogger().info(oauth_apps)

    auth_context = {}
    try:
        logging.getLogger().info("handler: Started Function Execution") 
        gateway_auth = json.loads(data.getvalue())
        auth_context = getAuthContext(gateway_auth['token'], oauth_apps)
        if (auth_context['active']):
            logging.getLogger().info('Authorizer returning 200...')
            return response.Response(
                ctx,
                response_data=json.dumps(auth_context),
                status_code = 200,
                headers={"Content-Type": "application/json"}
                )
        else:
            logging.getLogger().info('Authorizer returning 401...')
            return response.Response(
                ctx,
                response_data=json.dumps(str(auth_context)),
                status_code = 401,
                headers={"Content-Type": "application/json"}
                )

    except (Exception, ValueError) as ex:
        logging.getLogger().info('error parsing json payload: ' + str(ex))

        return response.Response(
            ctx,
            response_data=json.dumps(str(auth_context)),
            status_code = 500,
            headers={"Content-Type": "application/json"}
            )

    • 匯入

      • io、json、logging、datetime、time、base64：用於處理 I/O、JSON 資料、記錄日誌、日期與時間作業以及 base64 編碼的標準 Python 程式庫。
      • jwt：用於編碼和解碼 JSON Web 權杖 (JWT) 的程式庫。
      • requests：用於進行 HTTP 要求的程式庫。
      • HTTPBasicAuth：處理 HTTP 基本認證的類別。
      • serialization，default_backend：從用於處理加密作業的加密程式庫。
      • ociVault：用於與 OCI Vault 互動的自訂模組。

    • 全域變數

      • oauth_apps: 儲存應用程式組態的說明。

    • 功能

      • initContext(context): 此函數的目的是使用 OCI 保存庫的相關資訊環境資料和加密密碼來起始應用程式組態。它會接收相關資訊環境字典物件，該物件會被呼叫為主要處理程式方法中的第一件事，並使用任務 5 中說明的 getSecret() 函數從 OCI Vault 擷取加密密碼。

      • getAuthContext(token, client_apps): 植入並傳回 OCI API 閘道的認證相關資訊環境。擷取及解碼存取權杖。呼叫 introspectToken() 函數以驗證具有 Entra ID 的記號。如果權杖有效，請設定認證相關資訊環境，呼叫 getBackEndAuthToken() 函數以從 OCI IAM 擷取後端權杖並設定到期時間。如果記號無效，請設定 wwwAuthenticate 標頭來指示認證錯誤。

      • introspectToken(access_token, introspection_endpoint, client_id, client_secret): 使用提供的 introspection_endpoint 驗證記號。使用權杖對自我檢查端點提出 GET 要求。傳回自我檢查或驗證端點的回應。由於 Microsoft Entra ID 沒有 OAuth 自我檢查 API 端點，因此使用收到的權杖作為輸入項來呼叫設定的端點。

      • getBackEndAuthToken(token_endpoint, client_id, client_secret, scope, alias, principal): 從 PEM 檔案載入私密金鑰。建立 JWT 宣告並將它們編碼為 JWT 宣告。準備權杖要求的有效負載和標頭。對權杖端點提出 POST 要求，以取得後端權杖並將後端權杖傳回 getAuthContext() 函數。

      • handler(ctx, data: io.BytesIO = None): 處理函數執行的主要函數。使用 initContext() 函數起始 OAuth 相關資訊環境，並呼叫 getAuthContext() 函數以取得認證相關資訊環境。如果記號有效，則傳回 200 回應，否則會傳回 401 回應。記錄並傳回發生錯誤時的 500 回應。

工作 5：建立和設定 ociVault.py 檔案

在相同的資料夾中建立 ociVault.py 檔案，然後貼上下列程式碼片段。此公用程式功能會從 OCI 保存庫服務讀取加密密碼。

# Utility Function to get secrets from OCI Vault
import logging
import oci
import base64

def getSecret(ocid):
    signer = oci.auth.signers.get_resource_principals_signer()
    try:
        client = oci.secrets.SecretsClient({}, signer=signer)
        secret_content = client.get_secret_bundle(ocid).data.secret_bundle_content.content.encode('utf-8')
        decrypted_secret_content = base64.b64decode(secret_content).decode('utf-8')
    except Exception as ex:
        logging.getLogger().error("getSecret: Failed to get Secret" + ex)
        print("Error [getSecret]: failed to retrieve", ex, flush=True)
        raise
    return decrypted_secret_content

注意：請將工作 2 的 private_key.pem 檔案保留在相同的資料夾中。

影像 10

作業 5：測試函數

為了測試函數，我們需要部署函數，然後透過傳送 Microsoft Entra ID 權杖作為輸入來呼叫函數。

  1. 瀏覽至函數資料夾，然後執行下列命令 fn -v deploy --app MyCustomAuthorizer 來進行部署。Fn Project CLI 命令將建立函數，並在 OCI Functions 應用程式上部署相同函數。

    影像 11

    注意：在部署函數應用程式之前，請先在 requirements.txt 檔案中包含 fdk>=0.1.74、requests、oci、pyjwt、serialization。

  2. 使用 Postman 用戶端的 OAuth 2.0 ROPC 流程，從 Microsoft Entra ID 產生存取憑證。

    影像 12

  3. Note down the access token to generate a payload.json, that will be passed as an input to test the OCI Functions. Keep the JSON file in the same function directory.

    影像 13

  4. 儲存有效負載之後，您可以執行下列命令來模擬函數的執行，因為函數將透過 OCI API Gateway cat payload.json | fn invoke <AppName> <function name> 呼叫，如下圖所示。

    影像 14

    如果 Microsoft Entra ID 權杖有效，您將會看到一個回應，如下圖所示，您將會在相關資訊環境結構的 back_end_token 值中看到 OCI IAM 權杖值。

作業 6：設定 OCI API 閘道

OCI API Gateway 是一個完全託管、可擴展的雲端原生 API 管理平台，提供一系列服務，從快速 API 部署到生命週期管理和後端服務整合。我們將利用 API 閘道，使用外部身分識別提供者 (例如 Microsoft Entra ID) 來調解 Oracle Integration 的授權。

首先，請建立新的 API 閘道，然後在 API 閘道上建立新的部署。

  1. 瀏覽至 Developer Services 、 API Management 和 Gateways 。輸入下列資訊，然後按一下建立閘道。

    影像 15

    影像 16

  2. 在閘道詳細資訊頁面中，按一下建立部署，然後為您的 API 部署輸入下列必要資訊。

    • 名稱： 輸入名稱。
    • 路徑前置碼：定義路徑。
    • 區間： 為您的 API 部署選取適當的區間。

    影像 17

  3. 新增認證原則詳細資訊。您可以在此處設定要以自訂授權者方式呼叫的 OCI 函數。選取在任務 4 中建立的函數。

    影像 18

  4. 在路由頁面中，設定傳送至後端服務的 API 路由。在本教學課程中，我們將定義 Oracle Integration 端點的路由。

    影像 19

  5. 按一下顯示路由要求原則，使用者可以在此將認證權杖從 OCI 函數回應交換至要求的認證標頭。

    影像 20

    此步驟涉及根據後端身分識別提供者設定後端服務的認證權杖。在我們的案例中，我們正在設定從自訂授權者 OCI 函數收到的 OCI IAM Bearer 權杖。在這裡，我們將授權標頭設定為以 ${request.auth[back_end_token]} 值覆寫。請注意，back_end_token 是 Oracle 函數回應結構中相關資訊環境的一部分。請確定自訂授權者 OCI 函數完成後，此表示式已順利評估。

  6. 順利複查組態之後，請按一下儲存變更以儲存部署，並等待部署狀態變更為作用中。

    影像 21

    啟用 API 部署之後，請從部署資訊區段複製端點 (基礎 URL)。此 URL 是您部署的端點，您的業務流程或應用程式會使用 Microsoft Entra ID Bearer 權杖來呼叫 Oracle Integration 端點。我們將在下一個任務中使用基礎 URL。

    影像 22

作業 7：測試 API

首先，使用 Postman 用戶端從 Microsoft Entra ID 取得存取權杖。我們將使用 ROPC 流程來確保存取權杖包含必要的識別資訊。

  1. 複製存取記號，因為從 API 閘道呼叫 API 時將會使用相同的存取權杖。

    影像 12

  2. 建立結合從 API 閘道和 Oracle Integration 端點複製至作業 6 之基本端點 URL 的新 REST 要求，如以下影像所示。請在要求標頭中使用 Bearer 權杖。

    影像 23

  3. 按一下傳送即可呼叫 API 要求，它將會執行 Oracle Integration，並且應該提供成功的輸出。

    影像 24

接下來的步驟

我們已順利使用 Microsoft Entra ID 的 OAuth 權杖在 Oracle API Gateway 上呼叫 API，以接收來自 Oracle Integration REST 觸發程式流程的回應。這項整合對跨不同雲端供應商連接數位服務的客戶來說至關重要。

認可

其他學習資源

探索 docs.oracle.com/learn 上的其他實驗室，或存取 Oracle Learning YouTube 頻道上的更多免費學習內容。此外，請造訪 education.oracle.com/learning-explorer 以成為 Oracle Learning Explorer。

如需產品文件，請造訪 Oracle Help Center 。