boto3 我怎么能 cognito auth

Question

今天我想与 AWS Cognito 集成。我使用 Python SDK 接口 - boto3。

在文档中我可以找到注册帐户的方法，但我找不到身份验证用户。 文档：https://boto3.readthedocs.io/en/latest/reference/services/cognito-idp.html

我的问题是， 也许这个方法没有实现？所以，如果这个方法没有实现。也许有人为 AWS cognito 创建了一种身份验证方法？

谢谢大家:)

Answer 1

您需要编写自定义授权方。我使用无服务器来实现这一点，因为它提供了交叉编译在 lambda 上运行所需的本机库的能力。我创建了一个综合示例，应该对 here 有所帮助。

基础知识：

您将需要一些东西来验证令牌。我使用 python-jose：

def get_claims(event, context):
    token = event['authorizationToken'][7:]
    # get the kid from the headers prior to verification
    headers = jwt.get_unverified_headers(token)
    kid = headers['kid']
    # search for the kid in the downloaded public keys
    key_index = -1
    for i in range(len(keys)):
    if kid == keys[i]['kid']:
        key_index = i
        break
    if key_index == -1:
       print('Public key not found in jwks.json')
       return False
    # construct the public key
    public_key = jwk.construct(keys[key_index])
    # get the last two sections of the token,
    # message and signature (encoded in base64)
    message, encoded_signature = str(token).rsplit('.', 1)
    # decode the signature
    decoded_signature = base64url_decode(encoded_signature.encode('utf-8'))
    # verify the signature
    if not public_key.verify(message.encode("utf8"), decoded_signature):
      print('Signature verification failed')
      return False
    print('Signature successfully verified')
    # since we passed the verification, we can now safely
    # use the unverified claims
    claims = jwt.get_unverified_claims(token)
    # additionally we can verify the token expiration
    if time.time() > claims['exp']:
      print('Token is expired')
      return False
    # and the Audience  (use claims['client_id'] if verifying an access token)
    if 'aud' in claims and claims['aud'] != app_client_id:
      print('Token was not issued for this audience')
      return False
    # now we can use the claims
    return claimsenter code here

其次，您需要授权人根据声明返回策略，在此示例中允许所有路径，但如果您愿意，可以根据声明对其进行细化：

def authorize(event, context):
  print("Client token: " + event['authorizationToken'])
  print("Method ARN: " + event['methodArn'])

  """
  validate the incoming token
  and produce the principal user identifier associated with the token
  this could be accomplished in a number of ways:
  1. Call out to OAuth provider
  2. Decode a JWT token inline
  3. Lookup in a self-managed DB
  """

  token = event['authorizationToken'][7:]
  unverified_claims = jwt.get_unverified_claims(token)
  print json.dumps(unverified_claims)
  principalId = jwt.get_unverified_claims(token).get('username')

  """
  you can send a 401 Unauthorized response to the client by failing like so:
  raise Exception('Unauthorized')
  if the token is valid, a policy must be generated which will allow or deny access to the client
  if access is denied, the client will recieve a 403 Access Denied response
  if access is allowed, API Gateway will proceed with the backend integration configured on the method that was called
  this function must generate a policy that is associated with the recognized principal user identifier.
  depending on your use case, you might store policies in a DB, or generate them on the fly
  keep in mind, the policy is cached for 5 minutes by default (TTL is configurable in the authorizer)
  and will apply to subsequent calls to any method/resource in the RestApi
  made with the same token
  the example policy below denies access to all resources in the RestApi
  """

  tmp = event['methodArn'].split(':')
  apiGatewayArnTmp = tmp[5].split('/')
  awsAccountId = tmp[4]
  policy = AuthPolicy(principalId, awsAccountId)
  policy.restApiId = apiGatewayArnTmp[0]
  policy.region = tmp[3]
  policy.stage = apiGatewayArnTmp[1]
  try:
    print 'getting claims'
    #verified = verify_token(jwt_token,'access_token','access')
    claims = get_claims(event, context)
    print json.dumps(claims)
    if claims != False:
        print 'a'
        policy.allowAllMethods()
    else:
        policy.denyAllMethods()
  except:
    policy.denyAllMethods()
    """policy.allowMethod(HttpVerb.GET, "/pets/*")"""
    # Finally, build the policy
    authResponse = policy.build()
    # new! -- add additional key-value pairs associated with the authenticated principal
    # these are made available by APIGW like so: $context.authorizer.<key>
    # additional context is cached
    context = {
      'key': 'value', # $context.authorizer.key -> value
      'number' : 1,
      'bool' : True
    }
    # context['arr'] = ['foo'] <- this is invalid, APIGW will not accept it
    # context['obj'] = {'foo':'bar'} <- also invalid
    authResponse['context'] = context
    return authResponse

Answer 2

我们最近在我的授权 (https://github.com/capless/warrant/tree/develop) 项目中解决了这个问题。它在开发分支上，但将在本周晚些时候合并到 master 并发布到 pypi。

Answer 3

Cognito 用户池目前处于 Beta 版中，身份验证 API 目前还不是服务器端 SDK 的一部分。建议使用客户端 SDK 之一（Android、iOS 或 JavaScript）。

身份验证 API 将包含在服务器端 SDK 中，并提供该功能的一般可用性版本。