如何从aws上的python lambda中提取principalId

Question

我正在为我的 aws 无服务器应用程序使用 custom authorizer。

我有一个认知用户池。我要完成的工作流程是

1. 客户端登录到 cognito 用户池
2. 客户端获取令牌 id_token 以进行未来的 api 调用
3. 用户在我的 getUserInfo api 端点上发出 GET 请求，并传递令牌
4. lambda 函数使用令牌来验证 a) 用户是否已登录，以及 b) 用户的令牌是否与正在请求其信息的 user_name 匹配

我的理解是#4可以通过基于this aws documentation.查询context对象的authorizer.principalId属性来完成

但是，当我测试端点时，我在 python lambda 中的上下文对象不包含此授权者属性。

来自 CloudWatch 的 context 对象的属性转储如下：

{'aws_request_id': 'a37a1735-0ef3-*****-7fb1226218fe', 'log_group_name': '/aws/lambda/****', 'log_stream_name': '2019/01/03/[$LATEST]cca175ae1ff64cb699******', 'function_name': '*****getUser', 'memory_limit_in_mb': '1024', 'function_version': '$LATEST', 'invoked_function_arn': 'arn:aws:lambda:us-east-1:*********getUser', 'client_context': None, 'identity': <__main__.CognitoIdentity object at 0x7f1caf1bbdd8>}

我的 python lambda 格式为：

def getUser(event, context):
    print(context)
    print(context.__dict__)
    print(context.authorizer)
    print(context.identity.__dict__)
    print(event.__dict__)
    ...

如何让我的 lambda 检索所需的信息，以便我可以从传入的令牌验证用户名？

Answer 1

所需的授权信息可用于事件参数中的 lambda 函数。对于这种情况，我需要的是event['requestContext']['authorizer']['cognito:username']

Answer 2

下面的代码可能对你有帮助。

from __future__ import print_function
import boto3
import botocore.exceptions
import hmac
import hashlib
import base64
import json
import uuid
import logging
#Demo pool
USER_POOL_ID='ca-central-1_vaquarkhan'
CLIENT_ID ='vaquarkhanxxxxxxxxxxxxf7h8i9sc7ur6' # replace urs 
CLIENT_SECRET='vaquarkhanbryyyyyyyyyyyyyyyyhlk7ncv85qo5gjia1mdp9'# replace urs 


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

client = None

def get_secret_hash(username):
   msg = username + CLIENT_ID
   dig = hmac.new(str(CLIENT_SECRET).encode('utf-8'), 
   msg = str(msg).encode('utf-8'), digestmod=hashlib.sha256).digest()
   d2 = base64.b64encode(dig).decode()
 return d2

ERROR = 0
SUCCESS = 1
USER_EXISTS = 2


def sign_up(username,password):
   try:
     resp = client.sign_up(
         ClientId=CLIENT_ID ,
         SecretHash= get_secret_hash(username),
         Username=username,
         Password=password )
         print(resp)
         #except client.exceptions.UserNotFoundException as e:
         except client.exceptions.UsernameExistsException as e:
    return USER_EXISTS
    except Exception as e:
    print(e)
    logger.error(e)
    return ERROR
  return SUCCESS

def initiate_auth(username, password):
   try:
    # AdminInitiateAuth
      resp = client.admin_initiate_auth(
        UserPoolId=USER_POOL_ID,
        ClientId=CLIENT_ID,
        AuthFlow='ADMIN_NO_SRP_AUTH',
        AuthParameters={
            'USERNAME': username,
            'SECRET_HASH': get_secret_hash(username),
            'PASSWORD': password
          },
         ClientMetadata={
            'username': username,
            'password': password
           })
       except client.exceptions.NotAuthorizedException as e:
      return None, "The username or password incorrect"
    except client.exceptions.UserNotFoundException as e:
    return None, "Unauthorized"
    except Exception as e:
    print(e)
    logger.error(e)
    return None, "Unknown error"
return resp, None

def lambda_handler(event, context):
  global client
  if client == None:
    client = boto3.client('cognito-idp')

   print(event)
   body = event
   username = body['username']
   password = body['password']

   is_new ="false"
   user_id=str(uuid.uuid4())
  signed_up = sign_up(username,password)
  if signed_up == ERROR :
    return {'status':'fail','msg':'failed to sign up'}
  if signed_up == SUCCESS:
    is_new="true"

  resp,msg =initiate_auth(username,password)    

   if msg != None :
      logger.info('failed signIN with username={}'.format(username))
    return {'status':'fail','msg':msg}

id_token =resp['AuthenticationResult']['IdToken']
print('id token: '+id_token)
return {'status': 'success', 'id_token': id_token, 'user_id':user_id,'is_new':is_new}