Okta WorkflowsのLambdaコネクタを試してみた

Oktaでイベント発生（ユーザのディアクティベート）
    ↓
Okta Workflows（User Deactivated トリガー）
    ↓  AWS Lambda コネクタ
AWS Lambda（ユーザ情報の受け取り）
    ↓
CloudWatch Logs（ログ確認）

sam-app
├── samconfig.toml
├── src
│   └── lambda_function.py
└── template.yaml

version = 0.1

[default]
region = "ap-northeast-1"

[default.build.parameters]
debug = true

[default.deploy.parameters]
stack_name = "okta-workflows-test"
s3_bucket = "cm-da-uehara"
s3_prefix = "sam-deploy"
capabilities = "CAPABILITY_NAMED_IAM"
confirm_changeset = true

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: "SAM application for verifying the Okta Workflows AWS Lambda connector."

# ============================================================
# Globals
# ============================================================
Globals:
  Function:
    Runtime: python3.12
    Architectures:
      - x86_64
    Timeout: 30
    MemorySize: 128

# ============================================================
# Resources
# ============================================================
Resources:

  # ----------------------------------------------------------
  # Lambda function: retired user information receiver
  # ----------------------------------------------------------
  OktaRetirementReceiverFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: okta-retirement-receiver
      Description: >-
        Verification Lambda function that receives retired user information
        from the Okta Workflows Lambda connector and logs it to CloudWatch Logs.
      CodeUri: src/
      Handler: lambda_function.lambda_handler
      Role: !GetAtt LambdaExecutionRole.Arn

  # ----------------------------------------------------------
  # Lambda execution role
  # ----------------------------------------------------------
  LambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: okta-retirement-receiver-role
      Description: >-
        Execution role for the okta-retirement-receiver Lambda function.
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

  # ----------------------------------------------------------
  # IAM policy for Okta Workflows connectivity
  # ----------------------------------------------------------
  OktaWorkflowsInvokePolicy:
    Type: AWS::IAM::ManagedPolicy
    Properties:
      ManagedPolicyName: okta-workflows-invoke-retirement-receiver
      Description: >-
        Least-privilege policy for the Okta Workflows AWS Lambda connector.
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Sid: ListLambdaFunctions
            Effect: Allow
            Action: lambda:ListFunctions
            Resource: '*'
          - Sid: InvokeOktaRetirementReceiverFunction
            Effect: Allow
            Action: lambda:InvokeFunction
            Resource: !GetAtt OktaRetirementReceiverFunction.Arn

  # ----------------------------------------------------------
  # IAM user for Okta Workflows connectivity
  # (programmatic access only; access key must be issued manually after deployment)
  # ----------------------------------------------------------
  OktaWorkflowsLambdaInvokerUser:
    Type: AWS::IAM::User
    Properties:
      UserName: okta-workflows-lambda-invoker
      ManagedPolicyArns:
        - !Ref OktaWorkflowsInvokePolicy

# ============================================================
# Outputs
# ============================================================
Outputs:
  LambdaFunctionName:
    Description: Lambda function name
    Value: !Ref OktaRetirementReceiverFunction

  LambdaFunctionArn:
    Description: Lambda function ARN
    Value: !GetAtt OktaRetirementReceiverFunction.Arn

  IamUserName:
    Description: IAM user name for Okta Workflows connectivity
    Value: !Ref OktaWorkflowsLambdaInvokerUser

  IamPolicyArn:
    Description: IAM policy ARN for Okta Workflows connectivity
    Value: !Ref OktaWorkflowsInvokePolicy

from __future__ import annotations

import json
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def _extract_body(event: dict) -> dict:
    """受信イベントからリクエストボディを抽出する

    Lambda コネクター経由では event 自体が Payload となるが、
    API Gateway / Function URL 経由では event.body に JSON 文字列が包まれる形になる。
    どちらの呼び出し形式でも動作するよう両対応。

    Args:
        event (dict): Lambda が受け取るイベントオブジェクト

    Returns:
        dict: リクエストボディを表す辞書
    """
    if isinstance(event, dict) and isinstance(event.get("body"), str):
        try:
            return json.loads(event["body"])
        except json.JSONDecodeError:
            return {}
    return event if isinstance(event, dict) else {}

def _response(status_code: int, payload: dict) -> dict:
    """Lambda レスポンス辞書を生成する

    Args:
        status_code (int): HTTP ステータスコード
        payload (dict): レスポンスに含めるデータ

    Returns:
        dict: statusCode とペイロードをマージしたレスポンス辞書
    """
    return {"statusCode": status_code, **payload}

def lambda_handler(event: dict, context) -> dict:
    logger.info("受信 event: %s", json.dumps(event, ensure_ascii=False, default=str))

    body = _extract_body(event)

    email = (body.get("email") or "").strip()
    user_id = (body.get("user_id") or "").strip()
    display_name = (body.get("display_name") or "").strip()
    action = (body.get("action") or "").strip()

    logger.info(
        "退職者情報を受信しました / action=%s, email=%s, user_id=%s, display_name=%s",
        action,
        email,
        user_id,
        display_name,
    )

    result = {
        "message": "退職者情報を正常に受信しました",
        "received": {
            "action": action,
            "email": email,
            "user_id": user_id,
            "display_name": display_name,
        },
    }

    logger.info("処理完了: %s", json.dumps(result, ensure_ascii=False))
    return _response(200, {"result": result})

$ sam deploy