使用令牌验证 AWS Lambda

创建 Lambda 函数 ()AWS Management Console

1. 打开 Lamba 控制台的 Functions page（函数页面）。

2. 选择 Create function（创建函数）。

3. 选择从头开始编写。

4. 在基本信息下输入函数名称。在运行时下，选择基于 python 的运行时。

5. 在 Execution Role（执行角色）中，选择 Create a new role with basic Lambda permissions（创建具有基本 Lambda 权限的新角色）。

   Lambda 创建了一个执行角色，该角色向该函数授予将日志上传到亚马逊的权限。 CloudWatchLambda 函数在您调用函数时担任执行角色，并使用执行角色为 SDK 创建证书。 AWS

6. 选择代码选项卡，然后添加示例 Lambda 函数。

   如果您尚未为 Lambda 函数创建要使用的服务角色，则需要先创建该角色，然后示例 Lambda 函数才能运行。要了解有关为 Lambda 函数创建服务角色的更多信息，请参阅。创建用于解码 JWT 的 AWS Lambda 函数的 HealthLake 服务角色

   import base64
   import boto3
   import logging
   import json
   import os
   from urllib import request, parse
   
   logger = logging.getLogger()
   logger.setLevel(logging.INFO)
   
   ## Uses Secrets manager to gain access to the access key ID and secret access key for the authorization server
   client = boto3.client('secretsmanager', region_name="region-of-datastore")
   response = client.get_secret_value(SecretId='name-specified-by-customer-in-secretsmanager')
   secret = json.loads(response['SecretString'])
   client_id = secret['client_id']
   client_secret = secret['client_secret']
   
   
   unencoded_auth = f'{client_id}:{client_secret}'
   headers = {
     'Authorization': f'Basic {base64.b64encode(unencoded_auth.encode()).decode()}',
     'Content-Type': 'application/x-www-form-urlencoded'
   }
   
   auth_endpoint = os.environ['auth-server-base-url'] # Base URL of the Authorization server
   user_role_arn = os.environ['iam-role-arn'] # The IAM role client application will use to complete the HTTP request on the datastore
   
   def lambda_handler(event, context):
       if 'datastoreEndpoint' not in event or 'operationName' not in event or 'bearerToken' not in event:
       return {}
   
       datastore_endpoint = event['datastoreEndpoint']
       operation_name = event['operationName']
       bearer_token = event['bearerToken']
       logger.info('Datastore Endpoint [{}], Operation Name: [{}]'.format(datastore_endpoint, operation_name))
   
       ## To validate the token
       auth_response = auth_with_provider(bearer_token)
       logger.info('Auth response: [{}]'.format(auth_response))
       auth_payload = json.loads(auth_response)
       ## Required parameters needed to be sent to the datastore endpoint for the HTTP request to go through
       auth_payload["isAuthorized"] = bool(auth_payload["active"])
       auth_payload["nbf"] = auth_payload["iat"]
       return {"authPayload": auth_payload, "iamRoleARN": user_role_arn}
   
   ## Access the server
   def auth_with_provider(token):
       data = {'token': token, 'token_type_hint': 'access_token'}
       req = request.Request(url=auth_endpoint + '/v1/introspect', data=parse.urlencode(data).encode(), headers=headers)
       with request.urlopen(req) as resp:
       return resp.read().decode()