Oracle Cloud Infrastructureドキュメント

ファンクションの実行からのその他のOracle Cloud Infrastructureリソースへのアクセス

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

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

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

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

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. 動的グループのルールを指定する際は、次の例を考慮してください:

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

        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サービスと認証できるようにします。 参照:

例: 「ネットワーキング」サービスからVCNのリストを取得するためのPythonファンクションへのOracleリソース・プリンシパル・プロバイダの追加

ファンクションを動的グループに追加し、動的グループがテナンシ内のVCNをリストできるポリシーを作成しました。次の例のようなコードを使用すると、「ネットワーキング」サービスからVCNのリストを取得できます。 この例では、Oracleリソース・プリンシパルを使用して、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サービスにリクエストを作成する際に使用する秘密キー。

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

  • 2.2の値を含むOCI_RESOURCE_PRINCIPAL_VERSION。
  • rpstファイルへの絶対パスを含むOCI_RESOURCE_PRINCIPAL_RPST(ファイル名を含む)。
  • private.pemファイルへの絶対パスを含むOCI_RESOURCE_PRINCIPAL_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サービスのエンドポイント。
    • RPSTトークンのres_tenantおよびres_compartment要求を使用した、ファンクションのホスト・テナンシおよびコンパートメント。

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

#!/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 https://docs.cloud.oracle.com/iaas/Content/API/Concepts/signingrequests.htm#Python

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())