ファンクションの実行による他のOracle Cloud Infrastructureリソースへのアクセス

Oracle Functionsにデプロイしたファンクションが実行中の場合、他のOracle Cloud Infrastructureリソースにアクセスできます。例:

  • ファンクションでネットワーキング・サービスからVCNのリストを取得することが必要な場合があります。
  • ファンクションでオブジェクト・ストレージ・バケットからデータを読み取り、データに対してなんらかの操作を実行し、変更したデータをオブジェクト・ストレージ・バケットに書き戻すことが必要になる場合があります。

ファンクションが別のOracle Cloud Infrastructureリソースにアクセスできるようにするには、動的グループにファンクションを含めてから、動的グループにそのリソースへのアクセス権を付与するポリシーを作成する必要があります。動的グループの詳細とその作成に必要な権限などの詳細は、動的グループの管理を参照してください。

ポリシーと動的グループを設定することによって、ファンクションのコードに「リソース・プリンシパル・プロバイダ」へのコールを含めることができます。リソース・プリンシパル・プロバイダは、ファンクションが他のOracle Cloud Infrastructureサービスによって自身を認証できるようにするためのリソース・プロバイダ・セッション・トークン(RPST)を使用します。トークンは、動的グループがアクセス権を付与されているリソースに対してのみ有効です。

トークンは15分間キャッシュされることにも注意してください。そのため、ポリシーまたは動的グループを変更した場合、変更の影響を確認するには、15分待つ必要があります。

Oracleでは、Oracle Cloud Infrastructure SDKに含まれるリソース・プリンシパル・プロバイダを使用することをお薦めします。ただし、Oracle Cloud Infrastructure SDKでサポートされない言語でファンクションを記述している場合があります。または、単にOracle Cloud Infrastructure SDKを使用しない場合もあります。いずれの場合も、ファンクションが実行されているコンテナのファイルおよび環境変数を使用して、ファンクションが他のOracle Cloud Infrastructureサービスによって自身を認証できるようにする、カスタム・リソース・プリンシパル・プロバイダを作成できます。

コンソールの使用

実行中のファンクションが他のOracle Cloud Infrastructureリソースにアクセスできるようにするには:

  1. コンソールにログインして、新しい動的グループを作成します:

    1. ナビゲーション・メニューを開き、「アイデンティティとセキュリティ」をクリックします。「アイデンティティ」で、「動的グループ」をクリックします。
    2. 動的グループを作成するにはの手順に従い、動的グループに名前を付けます(たとえば、acme-func-dyn-grp)。
    3. 動的グループのルールを指定する場合は、次の例について検討してください:

      • コンパートメント内のすべてのファンクションがリソースにアクセスできるようにする場合は、次のようなルールを入力します。これにより、指定したコンパートメントOCIDがあるコンパートメント内のすべてのファンクションが動的グループに追加されます:

        ALL {resource.type = 'fnfunc', resource.compartment.id = 'ocid1.compartment.oc1..aaaaaaaa23______smwa'}
      • 特定のファンクションがリソースにアクセスできるようにする場合は、次のようなルールを入力します。これにより、指定したOCIDでファンクションが動的グループに追加されます:

        resource.id = 'ocid1.fnfunc.oc1.iad.aaaaaaaaacq______dnya'
      • 特定の定義済タグを持つすべてのファンクションがリソースにアクセスできるようにする場合は、次のようなルールを入力します。これにより、定義済のタグを持つすべてのファンクションが動的グループに追加されます:

        ALL {resource.type = 'fnfunc', tag.department.operations.value = '45'}

        フリーフォーム・タグはサポートされていません。タグ付けの詳細は、リソース・タグを参照してください。

    4. 「動的グループの作成」をクリックします。

    ファンクションを含む動的グループを作成したら、動的グループに必要なOracle Cloud Infrastructureリソースへのアクセス権を付与するポリシーを作成できます。

  2. 新規のポリシーの作成:

    1. ナビゲーション・メニューを開き、「アイデンティティとセキュリティ」をクリックします。「アイデンティティ」で、「ポリシー」をクリックします。
    2. ポリシーを作成するにはの手順に従い、ポリシーに名前を付けます(たとえば、acme-func-dyn-grp-policy)。
    3. ポリシー・ステートメントを指定する場合は、次の例について検討してください:

      • acme-func-dyn-grpのファンクションがテナンシ内のすべてのVCNのリストを取得できるようにする場合は、次のようなルールを入力します:

        allow dynamic-group acme-func-dyn-grp to inspect vcns in tenancy
      • acme-func-dyn-grp内のファンクションが特定のオブジェクト・ストレージ・バケットに対する読取りおよび書込みができるようにするには、次のようなルールを入力します:

        allow dynamic-group acme-func-dyn-grp to manage objects in compartment acme-storage-compartment where all {target.bucket.name='acme-functions-bucket'}
      • acme-func-dyn-grpのファンクションがコンパートメント内のすべてのリソースに対して読取りおよび書込みができるようにする場合は、次のようなルールを入力します。

        allow dynamic-group acme-func-dyn-grp to manage all-resources in compartment acme-storage-compartment
    4. 「作成」をクリックして、新しいポリシーを作成します。
  3. ファンクションが他のOracle Cloud Infrastructureサービスによって認証できるようにするには、リソース・プリンシパル・プロバイダをファンクション・コードに組み込みます。参照:

例: PythonファンクションへのOracle Resource Principal Providerの追加によるネットワーキング・サービスからのVCNのリストの取得

動的グループにファンクションを追加し、動的グループがテナンシ内のVCNをリストできるようにするポリシーを作成すると、次の例のようなコードを設定してネットワーキング・サービスからVCNのリストを取得できます。この例では、Oracle Resource Principal Providerを使用してRPSTトークンから資格証明を抽出します。

import io
import json

from fdk import response
import oci

def handler(ctx, data: io.BytesIO=None):
    signer = oci.auth.signers.get_resource_principals_signer()
    resp = do(signer)
    return response.Response(ctx,
        response_data=json.dumps(resp),
        headers={"Content-Type": "application/json"} )

def do(signer):
    # List VCNs --------------------------------------------------------
    client = oci.core.VirtualNetworkClient({}, signer=signer)
    try:
        vcns = client.list_vcns(signer.compartment_id)
        vcns = [[v.id, v.display_name] for v in vcns.data]
    except Exception as e:
        vcns = str(e)
    return {"vcns": vcns, }

例: ファンクションへのカスタム・リソース・プリンシパル・プロバイダの追加

Oracleでは、Oracle Cloud Infrastructure SDKに含まれるリソース・プリンシパル・プロバイダを使用することをお薦めします。ただし、Oracle Cloud Infrastructure SDKでサポートされない言語でファンクションを記述している場合があります。または、単にOracle Cloud Infrastructure SDKを使用しない場合もあります。いずれの場合も、ファンクションが実行されているコンテナのファイルおよび環境変数を使用して、ファンクションが他のOracle Cloud Infrastructureサービスによって自身を認証できるようにする、カスタム・リソース・プリンシパル・プロバイダを作成できます。

ファンクションが実行されるコンテナには、Oracle Cloud Infrastructureの互換性のある資格証明を保持するディレクトリ・ツリーが含まれます。具体的には次のとおりです:

  • rpstという名前のファイルにあるリソース・プリンシパル・セッション・トークン(RPST)。RPSTトークンはJWTトークンとして書式設定され、ファンクションのホスト・テナンシおよびコンパートメントを識別する要求を含みます。
  • private.pemという名前のファイルにある秘密キー。ファンクションのかわりにOracle Cloud Infrastructureサービスにリクエストを行うために使用されます。

次の環境変数は、ファンクションが実行されるコンテナ内で設定されます。

  • OCI_RESOURCE_PRINCIPAL_VERSION: 値2.2が含まれます。
  • OCI_RESOURCE_PRINCIPAL_RPST: rpstファイルへの絶対パスを含みます(ファイル名を含む)。
  • OCI_RESOURCE_PRINCIPAL_PRIVATE_PEM: private.pemファイルへの絶対パスを含みます(ファイル名を含む)。
  • OCI_RESOURCE_PRINCIPAL_REGION: ファンクションがデプロイされるリージョン識別子を含みます(たとえば、us-phoenix-1)。

ファンクションが別のOracle Cloud Infrastructureサービスにアクセスできるようにするには、ファンクションにコードを追加して、ファンクションが他のリソースにより認証できるようにします。

  1. OCI_RESOURCE_PRINCIPAL_RPST環境変数のパスからRPSTトークンをロードするコードを追加します。
  2. OCI_RESOURCE_PRINCIPAL_PRIVATE_PEM環境変数のパスから秘密キーをロードするコードを追加します。

  3. RPSTトークンと秘密キーを使用してOracle Cloud Infrastructureリクエストの署名を作成するコードを追加します(リクエストの署名を参照)。

  4. 他のOracle Cloud Infrastructureリソースへのリクエストを作成するコードを追加します。

    必要に応じて、次のものを識別できます:

    • OCI_RESOURCE_PRINCIPAL_REGION環境変数内のリージョン識別子を使用した、ファンクションと同じ(ローカル)リージョン内の他のOracle Cloud Infrastructureサービスのエンドポイント。
    • res_tenantおよびres_compartment要求をRPSTトークンに使用している、ファンクションのホスト・テナンシおよびコンパートメント。

たとえば、次のサンプルのPythonファンクションには、RPSTトークンから資格証明を抽出するカスタム・リソース・プリンシパル・プロバイダが含まれています。次に、IAM APIのgetTenancy操作にGETリクエストを発行して、ファンクションのテナンシのOCIDを戻します。

#!/usr/bin/env python3

import base64
import email.utils
import hashlib
import httpsig_cffi.sign
import json
import logging
import os.path
import re
import requests.auth
import urllib.parse


LOG = logging.getLogger(__name__)


# The following class is derived from the Python section in https://docs.cloud.oracle.com/iaas/Content/API/Concepts/signingrequests.htm

class SignedRequestAuth(requests.auth.AuthBase):
    """A requests auth instance that can be reused across requests"""
    generic_headers = [
        "date",
        "(request-target)",
        "host"
    ]
    body_headers = [
        "content-length",
        "content-type",
        "x-content-sha256",
    ]
    required_headers = {
        "get": generic_headers,
        "head": generic_headers,
        "delete": generic_headers,
        "put": generic_headers + body_headers,
        "post": generic_headers + body_headers,
    }

    def __init__(self, key_id, private_key):
        # Build a httpsig_cffi.requests_auth.HTTPSignatureAuth for each
        # HTTP method's required headers
        self.signers = {}
        for method, headers in self.required_headers.items():
            signer = httpsig_cffi.sign.HeaderSigner(
                key_id=key_id, secret=private_key,
                algorithm="rsa-sha256", headers=headers[:])
            use_host = "host" in headers
            self.signers[method] = (signer, use_host)

    def inject_missing_headers(self, request, sign_body):
        # Inject date, content-type, and host if missing
        request.headers.setdefault(
            "date", email.utils.formatdate(usegmt=True))
        request.headers.setdefault("content-type", "application/json")
        request.headers.setdefault(
            "host", urllib.parse.urlparse(request.url).netloc)

        # Requests with a body need to send content-type,
        # content-length, and x-content-sha256
        if sign_body:
            body = request.body or ""
            if "x-content-sha256" not in request.headers:
                m = hashlib.sha256(body.encode("utf-8"))
                base64digest = base64.b64encode(m.digest())
                base64string = base64digest.decode("utf-8")
                request.headers["x-content-sha256"] = base64string
            request.headers.setdefault("content-length", len(body))

    def __call__(self, request):
        verb = request.method.lower()
        # nothing to sign for options
        if verb == "options":
            return request
        signer, use_host = self.signers.get(verb, (None, None))
        if signer is None:
            raise ValueError(
                "Don't know how to sign request verb {}".format(verb))

        # Inject body headers for put/post requests, date for all requests
        sign_body = verb in ["put", "post"]
        self.inject_missing_headers(request, sign_body=sign_body)

        if use_host:
            host = urllib.parse.urlparse(request.url).netloc
        else:
            host = None

        signed_headers = signer.sign(
            request.headers, host=host,
            method=request.method, path=request.path_url)
        request.headers.update(signed_headers)
        return request


def rp_auther():
    if os.environ['OCI_RESOURCE_PRINCIPAL_VERSION'] != "2.2":
        raise EnvironmentError('{} must be set to the value "2.2"'.format('OCI_RESOURCE_PRINCIPAL_VERSION'))
    rpst = os.environ['OCI_RESOURCE_PRINCIPAL_RPST']
    if os.path.isabs(rpst):
        with open(rpst) as f:
            rpst = f.read()
    private_key = os.environ['OCI_RESOURCE_PRINCIPAL_PRIVATE_PEM']
    if os.path.isabs(private_key):
        with open(private_key) as f:
            private_key = f.read()
    return get_claims(rpst), SignedRequestAuth('ST${}'.format(rpst), private_key)


def get_claims(rpst):
    """Parse an RPST as a JWT; return a dictionary of claims

    The claims that are important are: sub, res_compartment, and res_tenant.
    These carry the resource OCID together with its location.
    """
    s = rpst.split('.')[1]
    s += "=" * ((4 - len(s) % 4) % 4)  # Pad to a multiple of 4 characters
    return json.loads(base64.b64decode(s).decode('utf-8'))


# Use RP credentials to make a request
region = os.environ['OCI_RESOURCE_PRINCIPAL_REGION']
claims, rp_auth = rp_auther()

response = requests.get("https://identity.{}.oraclecloud.com/20160918/tenancies/{}".format(region, claims['res_tenant']), auth=rp_auth)
print(response.json())